--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit ef977c84b347697d1f0b76858e17b3b7acae4109
Parents : 02ddb37
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-22T18:13:05-05:00
refactor: format and cleanup code using stricted ruff linter rules.
Changes
106 files changed, 593 insertions(+), 1339 deletions(-)
Diff
diff --git a/cx_setup.py b/cx_setup.py
index 694209f2..c502d9c4 100644
--- a/cx_setup.py
+++ b/cx_setup.py
@@ -20,9 +20,7 @@ changelog_path = ROOT / "CHANGELOG.md"
if changelog_path.exists():
include_files.append((str(changelog_path), "CHANGELOG.md"))
-frontend_licenses_path = (
- ROOT / "meshchatx" / "src" / "backend" / "data" / "licenses_frontend.json"
-)
+frontend_licenses_path = ROOT / "meshchatx" / "src" / "backend" / "data" / "licenses_frontend.json"
if frontend_licenses_path.exists():
include_files.append((str(frontend_licenses_path), "licenses_frontend.json"))
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index 9c340796..2183bbec 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -32,6 +32,7 @@ import webbrowser
import zipfile
from datetime import UTC, datetime, timedelta
from logging.handlers import RotatingFileHandler
+from typing import cast
from urllib.parse import urlparse
import aiohttp
@@ -47,6 +48,7 @@ from aiohttp_session.cookie_storage import EncryptedCookieStorage
from RNS.Discovery import InterfaceDiscovery
from serial.tools import list_ports
+from meshchatx.src.backend import gif_utils, sticker_pack_utils
from meshchatx.src.backend.announce_manager import (
filter_announced_dicts_by_search_query,
)
@@ -117,8 +119,6 @@ from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler
from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor
from meshchatx.src.backend.rnprobe_handler import RNProbeHandler
from meshchatx.src.backend.sideband_commands import SidebandCommands
-from meshchatx.src.backend import gif_utils
-from meshchatx.src.backend import sticker_pack_utils
from meshchatx.src.backend.sticker_utils import (
build_export_document,
mime_for_image_type,
@@ -426,9 +426,7 @@ class ReticulumMeshChat:
@property
def rnpath_trace_handler(self):
- return (
- self.current_context.rnpath_trace_handler if self.current_context else None
- )
+ return self.current_context.rnpath_trace_handler if self.current_context else None
@rnpath_trace_handler.setter
def rnpath_trace_handler(self, value):
@@ -473,11 +471,7 @@ class ReticulumMeshChat:
@property
def community_interfaces_manager(self):
- return (
- self.current_context.community_interfaces_manager
- if self.current_context
- else None
- )
+ return self.current_context.community_interfaces_manager if self.current_context else None
@community_interfaces_manager.setter
def community_interfaces_manager(self, value):
@@ -486,11 +480,7 @@ class ReticulumMeshChat:
@property
def local_lxmf_destination(self):
- return (
- self.current_context.local_lxmf_destination
- if self.current_context
- else None
- )
+ return self.current_context.local_lxmf_destination if self.current_context else None
@local_lxmf_destination.setter
def local_lxmf_destination(self, value):
@@ -505,11 +495,7 @@ class ReticulumMeshChat:
@property
def storage_path(self):
- return (
- self.current_context.storage_path
- if self.current_context
- else self.storage_dir
- )
+ return self.current_context.storage_path if self.current_context else self.storage_dir
@storage_path.setter
def storage_path(self, value):
@@ -828,10 +814,7 @@ class ReticulumMeshChat:
match = False
# check if local identity or destination matches
if hasattr(link, "destination") and link.destination:
- if (
- hasattr(link.destination, "identity")
- and link.destination.identity
- ):
+ if hasattr(link.destination, "identity") and link.destination.identity:
if link.destination.identity.hash == identity_hash_bytes:
match = True
@@ -839,7 +822,7 @@ class ReticulumMeshChat:
print(f"Tearing down RNS link {link}")
try:
link.teardown()
- except Exception: # noqa: S110
+ except Exception:
pass
except Exception as e:
print(f"Error while cleaning up RNS links: {e}")
@@ -938,7 +921,7 @@ class ReticulumMeshChat:
fileno = -1
try:
fileno = obj.fileno()
- except Exception: # noqa: S110
+ except Exception:
pass
with contextlib.suppress(Exception):
obj.close()
@@ -988,15 +971,10 @@ class ReticulumMeshChat:
else:
continue
- laddr_no_nul = (
- laddr_bytes[1:]
- if laddr_bytes.startswith(b"\0")
- else laddr_bytes
- )
+ laddr_no_nul = laddr_bytes[1:] if laddr_bytes.startswith(b"\0") else laddr_bytes
if (
- laddr_bytes == target_bytes
- or laddr_bytes == target_no_nul
+ laddr_bytes in (target_bytes, target_no_nul)
or laddr_no_nul == target_no_nul
):
try:
@@ -1009,7 +987,7 @@ class ReticulumMeshChat:
print(
f"Failed to close FD {fd} for {addr[1:]}: {fd_err}",
)
- except Exception: # noqa: S110
+ except Exception:
pass
except Exception as e:
print(f"Error scanning process for abstract UNIX FDs: {e}")
@@ -1118,7 +1096,7 @@ class ReticulumMeshChat:
try:
interface.server.shutdown()
interface.server.server_close()
- except Exception: # noqa: S110
+ except Exception:
pass
# AutoInterface specific
@@ -1127,7 +1105,7 @@ class ReticulumMeshChat:
try:
server.shutdown()
server.server_close()
- except Exception: # noqa: S110
+ except Exception:
pass
# For LocalServerInterface which Reticulum doesn't close properly
@@ -1135,7 +1113,7 @@ class ReticulumMeshChat:
try:
interface.server.shutdown()
interface.server.server_close()
- except Exception: # noqa: S110
+ except Exception:
pass
# TCPClientInterface/etc
@@ -1148,13 +1126,13 @@ class ReticulumMeshChat:
):
try:
interface.socket.shutdown(socket.SHUT_RDWR)
- except Exception: # noqa: S110
+ except Exception:
pass
try:
interface.socket.close()
- except Exception: # noqa: S110
+ except Exception:
pass
- except Exception: # noqa: S110
+ except Exception:
pass
interface.detach()
@@ -1213,7 +1191,7 @@ class ReticulumMeshChat:
try:
# Reticulum uses a staticmethod exit_handler
atexit.unregister(RNS.Reticulum.exit_handler)
- except Exception: # noqa: S110
+ except Exception:
pass
except Exception as e:
@@ -1261,13 +1239,9 @@ class ReticulumMeshChat:
)
# Only add if not already there
- if not any(
- addr == (rpc_bind, rpc_port) for addr, _ in rpc_addrs
- ):
+ if not any(addr == (rpc_bind, rpc_port) for addr, _ in rpc_addrs):
rpc_addrs.append(((rpc_bind, rpc_port), "AF_INET"))
- if not any(
- addr == (shared_bind, shared_port) for addr, _ in rpc_addrs
- ):
+ if not any(addr == (shared_bind, shared_port) for addr, _ in rpc_addrs):
rpc_addrs.append(((shared_bind, shared_port), "AF_INET"))
except Exception as e:
print(f"Warning reading Reticulum config for ports: {e}")
@@ -1282,11 +1256,7 @@ class ReticulumMeshChat:
all_free = True
for addr, family_str in rpc_addrs:
try:
- family = (
- socket.AF_INET
- if family_str == "AF_INET"
- else socket.AF_UNIX
- )
+ family = socket.AF_INET if family_str == "AF_INET" else socket.AF_UNIX
s = socket.socket(family, socket.SOCK_STREAM)
s.settimeout(0.5)
try:
@@ -1330,7 +1300,7 @@ class ReticulumMeshChat:
released = True
except OSError:
s2.close()
- except Exception: # noqa: S110
+ except Exception:
pass
if released:
@@ -1346,15 +1316,12 @@ class ReticulumMeshChat:
try:
current_process = psutil.Process()
# We use kind='all' to catch both TCP and UNIX sockets
- for conn in current_process.net_connections(
- kind="all"
- ):
+ for conn in current_process.net_connections(kind="all"):
try:
match = False
if conn.laddr:
- if (
- family_str == "AF_INET"
- and isinstance(conn.laddr, tuple)
+ if family_str == "AF_INET" and isinstance(
+ conn.laddr, tuple
):
# Match IP and port for IPv4
if conn.laddr.port == addr[1] and (
@@ -1390,15 +1357,13 @@ class ReticulumMeshChat:
target_addr.startswith(
b"\0",
)
- and current_laddr
- == target_addr[1:]
+ and current_laddr == target_addr[1:]
)
or (
current_laddr.startswith(
b"\0",
)
- and target_addr
- == current_laddr[1:]
+ and target_addr == current_laddr[1:]
)
):
match = True
@@ -1426,19 +1391,16 @@ class ReticulumMeshChat:
)
try:
- if (
- hasattr(conn, "fd")
- and conn.fd != -1
- ):
+ if hasattr(conn, "fd") and conn.fd != -1:
try:
os.close(conn.fd)
except Exception as fd_err:
print(
f"Failed to close FD {getattr(conn, 'fd', 'N/A')}: {fd_err}",
)
- except Exception: # noqa: S110
+ except Exception:
pass
- except Exception: # noqa: S110
+ except Exception:
pass
except Exception as e:
print(
@@ -1458,22 +1420,14 @@ class ReticulumMeshChat:
if not all_free:
await asyncio.sleep(2)
for addr, family_str in rpc_addrs:
- if (
- family_str == "AF_UNIX"
- and isinstance(addr, str)
- and addr.startswith("\0")
- ):
+ if family_str == "AF_UNIX" and isinstance(addr, str) and addr.startswith("\0"):
with contextlib.suppress(Exception):
self._force_close_abstract_unix_addr(addr)
last_check_all_free = True
for addr, family_str in rpc_addrs:
try:
- family = (
- socket.AF_INET
- if family_str == "AF_INET"
- else socket.AF_UNIX
- )
+ family = socket.AF_INET if family_str == "AF_INET" else socket.AF_UNIX
s = socket.socket(family, socket.SOCK_STREAM)
try:
s.bind(addr)
@@ -1490,9 +1444,9 @@ class ReticulumMeshChat:
continue
last_check_all_free = False
break
- except Exception: # noqa: S110
+ except Exception:
pass
- except Exception: # noqa: S110
+ except Exception:
pass
if not last_check_all_free:
@@ -1511,9 +1465,7 @@ class ReticulumMeshChat:
if abstract_unix_addr_in_use_after_wait:
original_instance_name = self._read_reticulum_instance_name()
base_name = original_instance_name or "default"
- switched_instance_name = (
- f"{base_name}-reload-{os.getpid()}-{int(time.time())}"
- )
+ switched_instance_name = f"{base_name}-reload-{os.getpid()}-{int(time.time())}"
self._write_reticulum_instance_name(switched_instance_name)
print(
"Abstract UNIX RPC address remained busy. "
@@ -1555,7 +1507,7 @@ class ReticulumMeshChat:
if not hasattr(self, "reticulum") and identity_to_restore is not None:
try:
self.setup_identity(identity_to_restore)
- except Exception: # noqa: S110
+ except Exception:
pass
return False
@@ -1608,9 +1560,7 @@ class ReticulumMeshChat:
"type": "identity_switched",
"identity_hash": identity_hash,
"display_name": (
- self.config.display_name.get()
- if hasattr(self, "config")
- else "Unknown"
+ self.config.display_name.get() if hasattr(self, "config") else "Unknown"
),
},
),
@@ -1685,9 +1635,7 @@ class ReticulumMeshChat:
def list_identities(self):
return self.identity_manager.list_identities(
- self.identity.hash.hex()
- if hasattr(self, "identity") and self.identity
- else None,
+ self.identity.hash.hex() if hasattr(self, "identity") and self.identity else None,
)
def create_identity(self, display_name=None):
@@ -1695,9 +1643,7 @@ class ReticulumMeshChat:
def delete_identity(self, identity_hash):
current_hash = (
- self.identity.hash.hex()
- if hasattr(self, "identity") and self.identity
- else None
+ self.identity.hash.hex() if hasattr(self, "identity") and self.identity else None
)
return self.identity_manager.delete_identity(identity_hash, current_hash)
@@ -1897,9 +1843,7 @@ class ReticulumMeshChat:
sanitized = []
seen = set()
for pattern in ReticulumMeshChat.parse_discovery_patterns(value):
- cleaned = (
- pattern.replace("\r", "").replace("\n", "").replace(",", "").strip()
- )
+ cleaned = pattern.replace("\r", "").replace("\n", "").replace(",", "").strip()
if not cleaned:
continue
cleaned = "".join(ch for ch in cleaned if ch.isprintable()).strip()
@@ -1947,11 +1891,7 @@ class ReticulumMeshChat:
or interface.get("remote")
or interface.get("listen_ip")
)
- port = (
- interface.get("port")
- or interface.get("target_port")
- or interface.get("listen_port")
- )
+ port = interface.get("port") or interface.get("target_port") or interface.get("listen_port")
if host and port:
candidates.append(f"{host}:{port}")
return candidates
@@ -1961,8 +1901,7 @@ class ReticulumMeshChat:
if not patterns:
return False
candidates = [
- value.lower()
- for value in ReticulumMeshChat.discovery_filter_candidates(interface)
+ value.lower() for value in ReticulumMeshChat.discovery_filter_candidates(interface)
]
for pattern in patterns:
normalized_pattern = str(pattern).lower()
@@ -2015,9 +1954,9 @@ class ReticulumMeshChat:
except Exception:
config_entry = None
- updated["ifac_netname"] = netname if netname else None
- updated["ifac_netkey"] = netkey if netkey else None
- updated["config_entry"] = config_entry if config_entry else None
+ updated["ifac_netname"] = netname or None
+ updated["ifac_netkey"] = netkey or None
+ updated["config_entry"] = config_entry or None
updated["network_name"] = updated["ifac_netname"]
updated["passphrase"] = updated["ifac_netkey"]
updated["publish_ifac"] = bool(
@@ -2040,10 +1979,7 @@ class ReticulumMeshChat:
interface
for interface in interfaces
if (
- (
- not whitelist
- or ReticulumMeshChat.matches_discovery_pattern(whitelist, interface)
- )
+ (not whitelist or ReticulumMeshChat.matches_discovery_pattern(whitelist, interface))
and not ReticulumMeshChat.matches_discovery_pattern(
blacklist,
interface,
@@ -2111,13 +2047,12 @@ class ReticulumMeshChat:
return
while self.running and ctx.running and ctx.session_id == session_id:
- auto_sync_interval_seconds = ctx.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
- last_synced_at = (
- ctx.config.lxmf_preferred_propagation_node_last_synced_at.get()
+ auto_sync_interval_seconds = (
+ ctx.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
)
+ last_synced_at = ctx.config.lxmf_preferred_propagation_node_last_synced_at.get()
should_sync = interval_action_due(
- auto_sync_interval_seconds is not None
- and auto_sync_interval_seconds > 0,
+ auto_sync_interval_seconds is not None and auto_sync_interval_seconds > 0,
last_synced_at,
auto_sync_interval_seconds,
time.time(),
@@ -2144,16 +2079,11 @@ class ReticulumMeshChat:
aspect="nomadnetwork.node",
)
for node in known_nodes:
- if (
- not self.running
- or not ctx.running
- or ctx.session_id != session_id
- ):
+ if not self.running or not ctx.running or ctx.session_id != session_id:
break
self.queue_crawler_task(
node["destination_hash"],
- ctx.config.nomad_default_page_path.get()
- or "/page/index.mu",
+ ctx.config.nomad_default_page_path.get() or "/page/index.mu",
context=ctx,
)
@@ -2167,10 +2097,7 @@ class ReticulumMeshChat:
# process tasks concurrently up to the limit
if tasks and self.running and ctx.running:
await asyncio.gather(
- *[
- self.process_crawler_task(task, context=ctx)
- for task in tasks
- ],
+ *[self.process_crawler_task(task, context=ctx) for task in tasks],
)
except Exception as e:
@@ -2343,8 +2270,8 @@ class ReticulumMeshChat:
metrics["started_at"] = datetime.now(UTC).isoformat()
metrics["baseline_total_messages"] = ctx.database.messages.count_lxmf_messages()
- metrics["baseline_delivered_messages"] = (
- ctx.database.messages.count_lxmf_messages_by_state("delivered")
+ metrics["baseline_delivered_messages"] = ctx.database.messages.count_lxmf_messages_by_state(
+ "delivered"
)
metrics["messages_stored"] = 0
metrics["delivery_confirmations"] = 0
@@ -2525,15 +2452,11 @@ class ReticulumMeshChat:
"rx_bytes": peer_rx_bytes + unpeered_rx_bytes,
"tx_bytes": peer_tx_bytes,
"unpeered_rx_bytes": unpeered_rx_bytes,
- "static_peers": stats.get("static_peers", 0)
- if isinstance(stats, dict)
- else 0,
+ "static_peers": stats.get("static_peers", 0) if isinstance(stats, dict) else 0,
"discovered_peers": (
stats.get("discovered_peers", 0) if isinstance(stats, dict) else 0
),
- "total_peers": stats.get("total_peers", 0)
- if isinstance(stats, dict)
- else 0,
+ "total_peers": stats.get("total_peers", 0) if isinstance(stats, dict) else 0,
"max_peers": stats.get("max_peers") if isinstance(stats, dict) else None,
"delivery_limit_bytes": int(delivery_limit * 1000),
"propagation_limit_bytes": int(propagation_limit * 1000),
@@ -2680,11 +2603,7 @@ class ReticulumMeshChat:
},
)
- if (
- hasattr(self, "reticulum")
- and self.reticulum
- and not self.reticulum.transport_enabled()
- ):
+ if hasattr(self, "reticulum") and self.reticulum and not self.reticulum.transport_enabled():
guidance.append(
{
"id": "transport_disabled",
@@ -2749,9 +2668,7 @@ class ReticulumMeshChat:
matches.add(message["source_hash"])
# also check custom display names
- custom_names = (
- self.database.announces.get_announces()
- ) # Or more specific if needed
+ custom_names = self.database.announces.get_announces() # Or more specific if needed
for announce in custom_names:
custom_name = self.database.announces.get_custom_display_name(
announce["destination_hash"],
@@ -2981,7 +2898,7 @@ class ReticulumMeshChat:
)
if contact:
target_name = contact.name
- except Exception: # noqa: S110
+ except Exception:
pass
AsyncUtils.run_async(
@@ -3017,7 +2934,7 @@ class ReticulumMeshChat:
continue
try:
ctx.teardown()
- except Exception: # noqa: S110
+ except Exception:
pass
self.contexts.clear()
self.current_context = None
@@ -3030,24 +2947,24 @@ class ReticulumMeshChat:
for websocket_client in list(self.websocket_clients):
try:
await websocket_client.close(code=WSCloseCode.GOING_AWAY)
- except Exception: # noqa: S110
+ except Exception:
pass
# stop reticulum
try:
RNS.Transport.detach_interfaces()
- except Exception: # noqa: S110
+ except Exception:
pass
if hasattr(self, "reticulum") and self.reticulum:
try:
self.reticulum.exit_handler()
- except Exception: # noqa: S110
+ except Exception:
pass
try:
RNS.exit()
- except Exception: # noqa: S110
+ except Exception:
pass
def exit_app(self, code=0):
@@ -3073,21 +2990,10 @@ class ReticulumMeshChat:
if not path.startswith("/api/"):
if (
path == "/"
- or path.startswith("/assets/")
- or path.startswith("/favicons/")
+ or path.startswith(("/assets/", "/favicons/"))
or path in ("/manifest.json", "/service-worker.js")
or path.endswith(
- (
- ".js",
- ".css",
- ".json",
- ".wasm",
- ".png",
- ".jpg",
- ".jpeg",
- ".ico",
- ".svg",
- ),
+ (".js", ".css", ".json", ".wasm", ".png", ".jpg", ".jpeg", ".ico", ".svg")
)
):
return await handler(request)
@@ -3118,17 +3024,10 @@ class ReticulumMeshChat:
# check if requesting setup page (index.html will show setup if needed)
if (
path == "/"
- or path.startswith("/assets/")
- or path.startswith("/favicons/")
- or path.endswith(".js")
- or path.endswith(".css")
- or path.endswith(".json")
- or path.endswith(".wasm")
- or path.endswith(".png")
- or path.endswith(".jpg")
- or path.endswith(".jpeg")
- or path.endswith(".ico")
- or path.endswith(".svg")
+ or path.startswith(("/assets/", "/favicons/"))
+ or path.endswith(
+ (".js", ".css", ".json", ".wasm", ".png", ".jpg", ".jpeg", ".ico", ".svg")
+ )
):
is_public = True
@@ -3509,8 +3408,7 @@ class ReticulumMeshChat:
return web.json_response(
{
"auth_enabled": self.auth_enabled,
- "password_set": self.config.auth_password_hash.get()
- is not None,
+ "password_set": self.config.auth_password_hash.get() is not None,
"authenticated": actually_authenticated,
},
)
@@ -3519,8 +3417,7 @@ class ReticulumMeshChat:
return web.json_response(
{
"auth_enabled": self.auth_enabled,
- "password_set": self.config.auth_password_hash.get()
- is not None,
+ "password_set": self.config.auth_password_hash.get() is not None,
"authenticated": False,
"error": str(e),
},
@@ -3775,9 +3672,7 @@ class ReticulumMeshChat:
async with session.get(url, allow_redirects=True) as response:
if response.status != 200:
return web.json_response(
- {
- "error": f"Failed to fetch release: {response.status}"
- },
+ {"error": f"Failed to fetch release: {response.status}"},
status=response.status,
)
data = await response.json(content_type=None)
@@ -4064,10 +3959,7 @@ class ReticulumMeshChat:
interface_details["type"] = interface_type
# if interface doesn't have enabled or interface_enabled setting already, enable it by default
- if (
- "enabled" not in interface_details
- and "interface_enabled" not in interface_details
- ):
+ if "enabled" not in interface_details and "interface_enabled" not in interface_details:
interface_details["interface_enabled"] = "true"
# handle AutoInterface
@@ -4099,8 +3991,7 @@ class ReticulumMeshChat:
return web.json_response(
{
"message": (
- "Multicast address type must be either "
- "'temporary' or 'permanent'"
+ "Multicast address type must be either 'temporary' or 'permanent'"
),
},
status=422,
@@ -4209,8 +4100,7 @@ class ReticulumMeshChat:
listen_ip_value = data.get("listen_ip")
listen_device_value = data.get("device")
if (listen_port_value not in (None, "")) and (
- listen_ip_value not in (None, "")
- or listen_device_value not in (None, "")
+ listen_ip_value not in (None, "") or listen_device_value not in (None, "")
):
if is_port_in_use(
listen_ip_value,
@@ -4255,10 +4145,7 @@ class ReticulumMeshChat:
status=422,
)
transport_identity = data.get("transport_identity")
- if (
- transport_identity is None
- or str(transport_identity).strip() == ""
- ):
+ if transport_identity is None or str(transport_identity).strip() == "":
return web.json_response(
{
"message": "Transport identity is required",
@@ -4279,8 +4166,7 @@ class ReticulumMeshChat:
else:
interface_details["connectable"] = (
"True"
- if str(connectable_value).lower()
- in {"true", "yes", "1", "on", "y"}
+ if str(connectable_value).lower() in {"true", "yes", "1", "on", "y"}
else "False"
)
peers = data.get("peers")
@@ -4289,9 +4175,7 @@ class ReticulumMeshChat:
cleaned_peers = [str(p).strip() for p in peers if str(p).strip()]
elif peers is not None and str(peers).strip() != "":
cleaned_peers = [
- s.strip()
- for s in str(peers).replace(",", " ").split()
- if s.strip()
+ s.strip() for s in str(peers).replace(",", " ").split() if s.strip()
]
if not cleaned_peers:
return web.json_response(
@@ -4730,10 +4614,7 @@ class ReticulumMeshChat:
interfaces = self._get_interfaces_snapshot()
for interface_name, interface in interfaces.items():
# skip interface if not selected
- if (
- selected_interface_names is not None
- and selected_interface_names != ""
- ):
+ if selected_interface_names is not None and selected_interface_names != "":
if interface_name not in selected_interface_names:
continue
@@ -4877,16 +4758,16 @@ class ReticulumMeshChat:
# handle websocket messages until disconnected
async for msg in websocket_response:
- msg: WSMessage = msg
- if msg.type == WSMsgType.TEXT:
+ message = cast(WSMessage, msg)
+ if message.type == WSMsgType.TEXT:
try:
- data = json.loads(msg.data)
+ data = json.loads(message.data)
await self.on_websocket_data_received(websocket_response, data)
except Exception as e:
# ignore errors while handling message
print("failed to process client message")
print(e)
- elif msg.type == WSMsgType.ERROR:
+ elif message.type == WSMsgType.ERROR:
# ignore errors while handling message
print(f"ws connection error {websocket_response.exception()}")
@@ -4920,12 +4801,12 @@ class ReticulumMeshChat:
)
async for msg in websocket_response:
- msg: WSMessage = msg
- if msg.type == WSMsgType.BINARY:
- self.web_audio_bridge.push_client_frame(msg.data)
- elif msg.type == WSMsgType.TEXT:
+ message = cast(WSMessage, msg)
+ if message.type == WSMsgType.BINARY:
+ self.web_audio_bridge.push_client_frame(message.data)
+ elif message.type == WSMsgType.TEXT:
try:
- data = json.loads(msg.data)
+ data = json.loads(message.data)
if data.get("type") == "attach":
self.web_audio_bridge.attach_client(websocket_response)
elif data.get("type") == "ping":
@@ -4936,7 +4817,7 @@ class ReticulumMeshChat:
logging.exception(
f"Error processing websocket text message: {e}",
)
- elif msg.type == WSMsgType.ERROR:
+ elif message.type == WSMsgType.ERROR:
print(f"telephone audio ws error {websocket_response.exception()}")
self.web_audio_bridge.detach_client(websocket_response)
@@ -5018,7 +4899,7 @@ class ReticulumMeshChat:
try:
path_table = self.reticulum.get_path_table()
total_paths = len(path_table)
- except Exception: # noqa: S110
+ except Exception:
pass
is_connected_to_shared_instance = getattr(
@@ -5033,24 +4914,17 @@ class ReticulumMeshChat:
for conn in process.net_connections(kind="all"):
if conn.status == psutil.CONN_ESTABLISHED and conn.raddr:
# Check for common Reticulum shared instance ports or UNIX sockets
- if (
- isinstance(conn.raddr, tuple)
- and conn.raddr[1] == 37428
- ):
- shared_instance_address = (
- f"{conn.raddr[0]}:{conn.raddr[1]}"
- )
+ if isinstance(conn.raddr, tuple) and conn.raddr[1] == 37428:
+ shared_instance_address = f"{conn.raddr[0]}:{conn.raddr[1]}"
break
if (
isinstance(conn.raddr, str)
- and (
- "rns" in conn.raddr or "reticulum" in conn.raddr
- )
+ and ("rns" in conn.raddr or "reticulum" in conn.raddr)
and ".sock" in conn.raddr
):
shared_instance_address = conn.raddr
break
- except Exception: # noqa: S110
+ except Exception:
pass
# Fallback to reading config if not found via connections
@@ -5075,10 +4949,8 @@ class ReticulumMeshChat:
"shared_instance_bind",
fallback="127.0.0.1",
)
- shared_instance_address = (
- f"{shared_bind}:{shared_port}"
- )
- except Exception: # noqa: S110
+ shared_instance_address = f"{shared_bind}:{shared_port}"
+ except Exception:
pass
# Calculate announce rates
@@ -5155,9 +5027,7 @@ class ReticulumMeshChat:
"database_file_size": db_files["main_bytes"],
"database_files": db_files,
"sqlite": {
- "journal_mode": _safe_sqlite_pragma(
- "journal_mode", "unknown"
- ),
+ "journal_mode": _safe_sqlite_pragma("journal_mode", "unknown"),
"synchronous": _safe_sqlite_pragma("synchronous", None),
"wal_autocheckpoint": _safe_sqlite_pragma(
"wal_autocheckpoint",
@@ -5203,8 +5073,7 @@ class ReticulumMeshChat:
[],
),
"user_guidance": _safe_user_guidance(),
- "tutorial_seen": _safe_config_get("tutorial_seen", "false")
- == "true",
+ "tutorial_seen": _safe_config_get("tutorial_seen", "false") == "true",
"changelog_seen_version": _safe_config_get(
"changelog_seen_version",
"0.0.0",
@@ -5413,9 +5282,7 @@ class ReticulumMeshChat:
async def docs_export(request):
try:
zip_data = self.docs_manager.export_docs()
- filename = (
- f"meshchatx_docs_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
- )
+ filename = f"meshchatx_docs_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.zip"
return web.Response(
body=zip_data,
content_type="application/zip",
@@ -5442,7 +5309,7 @@ class ReticulumMeshChat:
filename = (
"reticulum_manual_"
f"{safe_version}_"
- f"{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
+ f"{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.zip"
)
return web.Response(
body=zip_data,
@@ -5676,9 +5543,7 @@ class ReticulumMeshChat:
if self.database:
for item in identities:
if item.get("is_current"):
- item["message_count"] = (
- self.database.messages.count_lxmf_messages()
- )
+ item["message_count"] = self.database.messages.count_lxmf_messages()
break
return web.json_response(
{
@@ -6091,14 +5956,8 @@ class ReticulumMeshChat:
parsed_host = None
parsed_port = None
- host = (
- s.get("target_host") or s.get("remote") or parsed_host
- )
- port = (
- s.get("target_port")
- or s.get("listen_port")
- or parsed_port
- )
+ host = s.get("target_host") or s.get("remote") or parsed_host
+ port = s.get("target_port") or s.get("listen_port") or parsed_port
transport_id = s.get("transport_id")
if isinstance(transport_id, (bytes, bytearray)):
transport_id = transport_id.hex()
@@ -6133,10 +5992,8 @@ class ReticulumMeshChat:
return [to_jsonable(v) for v in obj]
return obj
- normalized_interfaces = (
- ReticulumMeshChat.normalize_discovered_ifac_fields(
- to_jsonable(interfaces),
- )
+ normalized_interfaces = ReticulumMeshChat.normalize_discovered_ifac_fields(
+ to_jsonable(interfaces),
)
return web.json_response(
@@ -6343,10 +6200,8 @@ class ReticulumMeshChat:
remote_identity = telephone_active_call.get_remote_identity()
if remote_identity:
caller_hash = remote_identity.hash.hex()
- contact = (
- self.database.contacts.get_contact_by_identity_hash(
- caller_hash,
- )
+ contact = self.database.contacts.get_contact_by_identity_hash(
+ caller_hash,
)
if not contact:
# Don't report active call if contacts-only is on and caller is not a contact
@@ -6433,7 +6288,7 @@ class ReticulumMeshChat:
)
if contact:
initiation_target_name = contact.name
- except Exception: # noqa: S110
+ except Exception:
pass
return web.json_response(
@@ -7068,16 +6923,12 @@ class ReticulumMeshChat:
ringtones = self.database.ringtones.get_all()
if ringtones:
- ringtone_id = random.choice(ringtones)["id"] # noqa: S311
+ ringtone_id = random.choice(ringtones)["id"]
else:
ringtone_id = None
has_custom = ringtone_id is not None
- ringtone = (
- self.database.ringtones.get_by_id(ringtone_id)
- if has_custom
- else None
- )
+ ringtone = self.database.ringtones.get_by_id(ringtone_id) if has_custom else None
return web.json_response(
{
@@ -7436,9 +7287,7 @@ class ReticulumMeshChat:
if sm is not None and sm > 0:
search_max = min(int(sm), 10_000)
- include_blocked = (
- request.query.get("include_blocked", "false").lower() == "true"
- )
+ include_blocked = request.query.get("include_blocked", "false").lower() == "true"
blocked_identity_hashes = None
if not include_blocked:
@@ -7501,7 +7350,7 @@ class ReticulumMeshChat:
def _fetch_custom_names():
return self.database.provider.fetchall(
- f"SELECT destination_hash, display_name FROM custom_destination_display_names WHERE destination_hash IN ({','.join(['?'] * len(other_user_hashes))})", # noqa: S608
+ f"SELECT destination_hash, display_name FROM custom_destination_display_names WHERE destination_hash IN ({','.join(['?'] * len(other_user_hashes))})",
other_user_hashes,
)
@@ -7512,19 +7361,13 @@ class ReticulumMeshChat:
# If we're looking for telephony announces, pre-fetch LXMF announces for the same identities
if aspect == "lxst.telephony":
identity_hashes = list(
- set(
- [
- r["identity_hash"]
- for r in results
- if r.get("identity_hash")
- ],
- ),
+ {r["identity_hash"] for r in results if r.get("identity_hash")},
)
if identity_hashes:
def _fetch_lxmf_names():
return self.database.announces.provider.fetchall(
- f"SELECT identity_hash, app_data FROM announces WHERE aspect = 'lxmf.delivery' AND identity_hash IN ({','.join(['?'] * len(identity_hashes))})", # noqa: S608
+ f"SELECT identity_hash, app_data FROM announces WHERE aspect = 'lxmf.delivery' AND identity_hash IN ({','.join(['?'] * len(identity_hashes))})",
identity_hashes,
)
@@ -7642,9 +7485,7 @@ class ReticulumMeshChat:
results = self.database.announces.get_favourites(aspect=aspect)
# process favourites
- favourites = [
- convert_db_favourite_to_dict(favourite) for favourite in results
- ]
+ favourites = [convert_db_favourite_to_dict(favourite) for favourite in results]
return web.json_response(
{
@@ -7839,9 +7680,7 @@ class ReticulumMeshChat:
* 100, # convert to percentage
"messages_received": self.message_router.propagation_transfer_last_result,
"messages_stored": sync_metrics["messages_stored"],
- "delivery_confirmations": sync_metrics[
- "delivery_confirmations"
- ],
+ "delivery_confirmations": sync_metrics["delivery_confirmations"],
"messages_hidden": sync_metrics["messages_hidden"],
},
"local_propagation_node": self.get_local_propagation_node_stats(),
@@ -7948,25 +7787,19 @@ class ReticulumMeshChat:
local_destination_hash = local_destination_hash_raw
else:
local_destination_hash = None
- local_stats = (
- self.get_local_propagation_node_stats(context=ctx) if ctx else None
- )
+ local_stats = self.get_local_propagation_node_stats(context=ctx) if ctx else None
for announce in results:
# find an lxmf.delivery announce for the same identity hash, so we can use that as an "operater by" name
lxmf_delivery_results = self.database.announces.get_filtered_announces(
aspect="lxmf.delivery",
identity_hash=announce["identity_hash"],
)
- lxmf_delivery_announce = (
- lxmf_delivery_results[0] if lxmf_delivery_results else None
- )
+ lxmf_delivery_announce = lxmf_delivery_results[0] if lxmf_delivery_results else None
# find a nomadnetwork.node announce for the same identity hash, so we can use that as an "operated by" name
- nomadnetwork_node_results = (
- self.database.announces.get_filtered_announces(
- aspect="nomadnetwork.node",
- identity_hash=announce["identity_hash"],
- )
+ nomadnetwork_node_results = self.database.announces.get_filtered_announces(
+ aspect="nomadnetwork.node",
+ identity_hash=announce["identity_hash"],
)
nomadnetwork_node_announce = (
nomadnetwork_node_results[0] if nomadnetwork_node_results else None
@@ -8055,9 +7888,7 @@ class ReticulumMeshChat:
else ctx.config.lxmf_local_propagation_node_enabled.get()
),
"per_transfer_limit": int(
- getattr(
- ctx.message_router, "propagation_per_transfer_limit", 0
- ),
+ getattr(ctx.message_router, "propagation_per_transfer_limit", 0),
),
"is_local_node": True,
"local_node_stats": local_stats,
@@ -8264,8 +8095,7 @@ class ReticulumMeshChat:
# get signal metrics from latest lxmf message if it's more recent than the announce
if latest_lxmf_message is not None and (
- latest_announce_at is None
- or latest_lxmf_message_at > latest_announce_at
+ latest_announce_at is None or latest_lxmf_message_at > latest_announce_at
):
snr = latest_lxmf_message["snr"]
rssi = latest_lxmf_message["rssi"]
@@ -8307,8 +8137,7 @@ class ReticulumMeshChat:
# wait until we have a path, or give up after the configured timeout
while (
- not RNS.Transport.has_path(destination_hash)
- and time.time() < timeout_after_seconds
+ not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after_seconds
):
await asyncio.sleep(0.1)
@@ -8998,9 +8827,7 @@ class ReticulumMeshChat:
if ts is not None and hasattr(ts, "isoformat"):
bot["last_announce_at"] = ts.isoformat()
else:
- bot["last_announce_at"] = (
- str(ts) if ts is not None else None
- )
+ bot["last_announce_at"] = str(ts) if ts is not None else None
return web.json_response(
{
"status": status,
@@ -9250,8 +9077,8 @@ class ReticulumMeshChat:
)
# get outbound ticket expiry for this lxmf destination
- lxmf_outbound_ticket_expiry = (
- self.message_router.get_outbound_ticket_expiry(destination_hash_bytes)
+ lxmf_outbound_ticket_expiry = self.message_router.get_outbound_ticket_expiry(
+ destination_hash_bytes
)
return web.json_response(
@@ -9274,9 +9101,7 @@ class ReticulumMeshChat:
# ensure transport_id is hex as json_response can't serialize bytes
if "transport_id" in interface_stats:
- interface_stats["transport_id"] = interface_stats[
- "transport_id"
- ].hex()
+ interface_stats["transport_id"] = interface_stats["transport_id"].hex()
# ensure probe_responder is hex as json_response can't serialize bytes
if (
@@ -9301,16 +9126,14 @@ class ReticulumMeshChat:
].hex()
if interface.get("ifac_signature"):
- interface["ifac_signature"] = interface[
- "ifac_signature"
- ].hex()
+ interface["ifac_signature"] = interface["ifac_signature"].hex()
try:
if interface.get("hash"):
interface["hash"] = interface["hash"].hex()
- except Exception: # noqa: S110
+ except Exception:
pass
- except Exception: # noqa: S110
+ except Exception:
pass
return web.json_response(
@@ -9332,23 +9155,19 @@ class ReticulumMeshChat:
destination_hashes = body.get("destination_hashes")
if destination_hashes and not isinstance(destination_hashes, list):
destination_hashes = None
- except Exception: # noqa: S110
+ except Exception:
pass
all_paths = []
if hasattr(self, "reticulum") and self.reticulum:
try:
all_paths = self.reticulum.get_path_table()
- except Exception: # noqa: S110
+ except Exception:
pass
if destination_hashes:
- hash_set = set(
- h.lower() for h in destination_hashes if isinstance(h, str)
- )
- all_paths = [
- p for p in all_paths if p["hash"].hex().lower() in hash_set
- ]
+ hash_set = {h.lower() for h in destination_hashes if isinstance(h, str)}
+ all_paths = [p for p in all_paths if p["hash"].hex().lower() in hash_set]
total_count = len(all_paths)
@@ -9416,9 +9235,7 @@ class ReticulumMeshChat:
file_attachments_field = None
if "file_attachments" in fields:
file_attachments = []
- for file_attachment in data["lxmf_message"]["fields"][
- "file_attachments"
- ]:
+ for file_attachment in data["lxmf_message"]["fields"]["file_attachments"]:
file_name = file_attachment["file_name"]
file_bytes = base64.b64decode(file_attachment["file_bytes"])
file_attachments.append(LxmfFileAttachment(file_name, file_bytes))
@@ -9457,9 +9274,7 @@ class ReticulumMeshChat:
reply_to_hash = None
if "reply_to_hash" in data["lxmf_message"]:
reply_to_hash = data["lxmf_message"]["reply_to_hash"]
- reply_quoted_content = (
- data["lxmf_message"].get("reply_quoted_content") or None
- )
+ reply_quoted_content = data["lxmf_message"].get("reply_quoted_content") or None
try:
# send lxmf message to destination
@@ -9630,8 +9445,7 @@ class ReticulumMeshChat:
# convert to response json
lxmf_messages = [
- convert_db_lxmf_message_to_dict(db_lxmf_message)
- for db_lxmf_message in results
+ convert_db_lxmf_message_to_dict(db_lxmf_message) for db_lxmf_message in results
]
return web.json_response(
@@ -9768,9 +9582,7 @@ class ReticulumMeshChat:
data = await request.json()
except Exception:
return web.json_response({"message": "invalid json"}, status=400)
- destination_hash = (
- data.get("destination_hash") if isinstance(data, dict) else None
- )
+ destination_hash = data.get("destination_hash") if isinstance(data, dict) else None
if not destination_hash:
return web.json_response(
{"message": "missing destination_hash"},
@@ -9866,9 +9678,7 @@ class ReticulumMeshChat:
}
# contact image
- contact_image = (
- row["contact_image"] if "contact_image" in row.keys() else None
- )
+ contact_image = row.get("contact_image", None)
is_unread = compute_lxmf_conversation_unread_from_latest_row(row)
@@ -9997,9 +9807,7 @@ class ReticulumMeshChat:
@routes.get("/api/v1/lxmf/folders/export")
async def lxmf_folders_export(request):
folders = [dict(f) for f in self.database.messages.get_all_folders()]
- mappings = [
- dict(m) for m in self.database.messages.get_all_conversation_folders()
- ]
+ mappings = [dict(m) for m in self.database.messages.get_all_conversation_folders()]
return web.json_response({"folders": folders, "mappings": mappings})
@routes.post("/api/v1/lxmf/folders/import")
@@ -10128,8 +9936,10 @@ class ReticulumMeshChat:
db_message.get("content"),
db_message.get("title"),
):
- latest_user_facing = self.database.messages.get_latest_user_facing_incoming_message(
- other_user_hash,
+ latest_user_facing = (
+ self.database.messages.get_latest_user_facing_incoming_message(
+ other_user_hash,
+ )
)
if latest_user_facing is None:
continue
@@ -10144,10 +9954,7 @@ class ReticulumMeshChat:
last_read_dt = last_read_dt.replace(
tzinfo=UTC,
)
- if (
- latest_user_facing["timestamp"]
- <= last_read_dt.timestamp()
- ):
+ if latest_user_facing["timestamp"] <= last_read_dt.timestamp():
continue
except (ValueError, TypeError):
pass
@@ -10166,10 +9973,8 @@ class ReticulumMeshChat:
display_name = self.get_lxmf_conversation_name(
other_user_hash,
)
- custom_display_name = (
- self.database.announces.get_custom_display_name(
- other_user_hash,
- )
+ custom_display_name = self.database.announces.get_custom_display_name(
+ other_user_hash,
)
# Determine latest message data
@@ -10188,9 +9993,9 @@ class ReticulumMeshChat:
"display_name": display_name,
"custom_display_name": custom_display_name,
"lxmf_user_icon": dict(icon) if icon else None,
- "latest_message_preview": (
- latest_message_data["content"] or ""
- )[:100],
+ "latest_message_preview": (latest_message_data["content"] or "")[
+ :100
+ ],
"updated_at": datetime.fromtimestamp(
latest_message_data["timestamp"] or 0,
UTC,
@@ -10280,8 +10085,10 @@ class ReticulumMeshChat:
conv.get("content"),
conv.get("title"),
):
- latest_user_facing = self.database.messages.get_latest_user_facing_incoming_message(
- other_user_hash,
+ latest_user_facing = (
+ self.database.messages.get_latest_user_facing_incoming_message(
+ other_user_hash,
+ )
)
if latest_user_facing is None:
continue
@@ -10295,10 +10102,7 @@ class ReticulumMeshChat:
last_read_dt = last_read_dt.replace(
tzinfo=UTC,
)
- if (
- latest_user_facing["timestamp"]
- <= last_read_dt.timestamp()
- ):
+ if latest_user_facing["timestamp"] <= last_read_dt.timestamp():
continue
except (ValueError, TypeError):
pass
@@ -10456,9 +10260,7 @@ class ReticulumMeshChat:
formatted = {}
for h, info in identities.items():
formatted[h.hex()] = {
- "source": info.get("source", b"").hex()
- if info.get("source")
- else None,
+ "source": info.get("source", b"").hex() if info.get("source") else None,
"until": info.get("until"),
"reason": info.get("reason"),
}
@@ -10908,9 +10710,7 @@ class ReticulumMeshChat:
pack_id = int(request.match_info.get("pack_id", "0"))
except ValueError:
return web.json_response({"error": "invalid_pack_id"}, status=400)
- with_stickers = (
- request.query.get("with_stickers", "false").lower() == "true"
- )
+ with_stickers = request.query.get("with_stickers", "false").lower() == "true"
if with_stickers:
ok = self.database.sticker_packs.delete_with_stickers(
pack_id,
@@ -11179,9 +10979,7 @@ class ReticulumMeshChat:
"destination_hash": r["destination_hash"],
"timestamp": r["timestamp"],
"telemetry": unpacked,
- "physical_link": json.loads(r["physical_link"])
- if r["physical_link"]
- else None,
+ "physical_link": json.loads(r["physical_link"]) if r["physical_link"] else None,
"updated_at": r["updated_at"],
},
)
@@ -11321,10 +11119,8 @@ class ReticulumMeshChat:
path = request.path
if path.startswith("/api/"):
return response
- if path.endswith(".js") or path.endswith(".mjs"):
- response.headers["Content-Type"] = (
- "application/javascript; charset=utf-8"
- )
+ if path.endswith((".js", ".mjs")):
+ response.headers["Content-Type"] = "application/javascript; charset=utf-8"
elif path.endswith(".css"):
response.headers["Content-Type"] = "text/css; charset=utf-8"
elif path.endswith(".json"):
@@ -11407,7 +11203,7 @@ class ReticulumMeshChat:
if domain not in target_list:
target_list.append(domain)
return domain
- except Exception: # noqa: S110
+ except Exception:
pass
return None
@@ -11432,9 +11228,7 @@ class ReticulumMeshChat:
return
sources = [
s.strip()
- for s in extra_str.replace("\n", ",")
- .replace(";", ",")
- .split(",")
+ for s in extra_str.replace("\n", ",").replace(";", ",").split(",")
if s.strip()
]
for s in sources:
@@ -11573,9 +11367,7 @@ class ReticulumMeshChat:
def run(self, host, port, launch_browser: bool, enable_https: bool = True):
# create route table
routes = web.RouteTableDef()
- auth_middleware, mime_type_middleware, security_middleware = (
- self._define_routes(routes)
- )
+ auth_middleware, mime_type_middleware, security_middleware = self._define_routes(routes)
ssl_context = None
use_https = enable_https
@@ -11846,9 +11638,7 @@ class ReticulumMeshChat:
# Local node selected as preferred: no transport path lookup is needed.
# Mark sync as complete immediately to avoid getting stuck in PR_PATH_REQUESTED.
with contextlib.suppress(Exception):
- ctx.message_router.propagation_transfer_state = (
- ctx.message_router.PR_COMPLETE
- )
+ ctx.message_router.propagation_transfer_state = ctx.message_router.PR_COMPLETE
ctx.message_router.propagation_transfer_progress = 1.0
ctx.message_router.propagation_transfer_last_result = 0
await self.send_config_to_websocket_clients(context=ctx)
@@ -13086,9 +12876,7 @@ class ReticulumMeshChat:
bytes.fromhex(destination_hash_hex)
raw_bytes = bytes.fromhex(public_key_hex)
- public_key_bytes = (
- raw_bytes[:32] if len(raw_bytes) >= 32 else raw_bytes
- )
+ public_key_bytes = raw_bytes[:32] if len(raw_bytes) >= 32 else raw_bytes
identity = RNS.Identity(create_keys=False)
if not identity.load_public_key(public_key_bytes):
@@ -13099,10 +12887,8 @@ class ReticulumMeshChat:
raise ValueError("Invalid LXMA public key")
remote_identity_hash = identity.hash.hex()
- existing_contact = (
- self.database.contacts.get_contact_by_identity_hash(
- remote_identity_hash,
- )
+ existing_contact = self.database.contacts.get_contact_by_identity_hash(
+ remote_identity_hash,
)
contact_name = (
existing_contact["name"]
@@ -13691,9 +13477,9 @@ class ReticulumMeshChat:
"lxmf",
"delivery",
).hex()
- except Exception: # noqa: S110
+ except Exception:
pass
- except Exception: # noqa: S110
+ except Exception:
pass
# find lxmf user icon from database
@@ -13774,10 +13560,7 @@ class ReticulumMeshChat:
# ensure we're not storing the user's own icon with a peer's hash
# only store icons for remote peers, not for the local user
- if (
- ctx.local_lxmf_destination
- and destination_hash == ctx.local_lxmf_destination.hexhash
- ):
+ if ctx.local_lxmf_destination and destination_hash == ctx.local_lxmf_destination.hexhash:
print(f"skipping icon update for local user's own hash: {destination_hash}")
return
@@ -13901,8 +13684,7 @@ class ReticulumMeshChat:
and (
SidebandCommands.TELEMETRY_REQUEST in command
or str(SidebandCommands.TELEMETRY_REQUEST) in command
- or f"0x{SidebandCommands.TELEMETRY_REQUEST:02x}"
- in command
+ or f"0x{SidebandCommands.TELEMETRY_REQUEST:02x}" in command
)
)
or (
@@ -13978,9 +13760,7 @@ class ReticulumMeshChat:
# check for spam keywords
is_spam = False
message_title = lxmf_message.title if hasattr(lxmf_message, "title") else ""
- message_content = (
- lxmf_message.content if hasattr(lxmf_message, "content") else ""
- )
+ message_content = lxmf_message.content if hasattr(lxmf_message, "content") else ""
if isinstance(message_content, bytes):
message_content = message_content.decode("utf-8", errors="replace")
elif message_content is None:
@@ -14017,9 +13797,8 @@ class ReticulumMeshChat:
)
return
# strip attachments from strangers (non-contacts) if setting is enabled
- if (
- ctx.config.block_attachments_from_strangers.get()
- and not self._is_contact(source_hash, context=ctx)
+ if ctx.config.block_attachments_from_strangers.get() and not self._is_contact(
+ source_hash, context=ctx
):
for key in (
LXMF.FIELD_FILE_ATTACHMENTS,
@@ -14068,9 +13847,7 @@ class ReticulumMeshChat:
for entry in stream:
if isinstance(entry, (list, tuple)) and len(entry) >= 3:
entry_source = (
- entry[0].hex()
- if isinstance(entry[0], bytes)
- else entry[0]
+ entry[0].hex() if isinstance(entry[0], bytes) else entry[0]
)
entry_timestamp = entry[1]
entry_data = entry[2]
@@ -14094,9 +13871,7 @@ class ReticulumMeshChat:
background_colour = "#" + icon_appearance[2].hex()
local_hash = (
- ctx.local_lxmf_destination.hexhash
- if ctx.local_lxmf_destination
- else None
+ ctx.local_lxmf_destination.hexhash if ctx.local_lxmf_destination else None
)
source_hash = lxmf_message.source_hash.hex()
@@ -14107,12 +13882,8 @@ class ReticulumMeshChat:
pass
else:
local_icon_name = ctx.config.lxmf_user_icon_name.get()
- local_icon_fg = (
- ctx.config.lxmf_user_icon_foreground_colour.get()
- )
- local_icon_bg = (
- ctx.config.lxmf_user_icon_background_colour.get()
- )
+ local_icon_fg = ctx.config.lxmf_user_icon_foreground_colour.get()
+ local_icon_bg = ctx.config.lxmf_user_icon_background_colour.get()
# if incoming icon matches our own, skip storing and clear any mistaken stored copy
# for now, but this will need to be updated later if two users do have the same icon
@@ -14223,9 +13994,7 @@ class ReticulumMeshChat:
self.send_message(
destination_hash=mapping["original_sender_hash"],
content=lxmf_message.content,
- title=lxmf_message.title
- if hasattr(lxmf_message, "title")
- else "",
+ title=lxmf_message.title if hasattr(lxmf_message, "title") else "",
image_field=image_field,
audio_field=audio_field,
file_attachments_field=file_attachments_field,
@@ -14244,10 +14013,7 @@ class ReticulumMeshChat:
for rule in rules:
# check source filter if set
- if (
- rule["source_filter_hash"]
- and rule["source_filter_hash"] != source_hash
- ):
+ if rule["source_filter_hash"] and rule["source_filter_hash"] != source_hash:
continue
# find or create mapping for this (Source, Final Recipient) pair
@@ -14265,9 +14031,7 @@ class ReticulumMeshChat:
self.send_message(
destination_hash=rule["forward_to_hash"],
content=lxmf_message.content,
- title=lxmf_message.title
- if hasattr(lxmf_message, "title")
- else "",
+ title=lxmf_message.title if hasattr(lxmf_message, "title") else "",
sender_identity_hash=mapping["alias_hash"],
image_field=image_field,
audio_field=audio_field,
@@ -14366,10 +14130,7 @@ class ReticulumMeshChat:
# resend message
source_hash = lxmf_message.source_hash.hex()
router = ctx.message_router
- if (
- ctx.forwarding_manager
- and source_hash in ctx.forwarding_manager.forwarding_routers
- ):
+ if ctx.forwarding_manager and source_hash in ctx.forwarding_manager.forwarding_routers:
router = ctx.forwarding_manager.forwarding_routers[source_hash]
router.handle_outbound(lxmf_message)
@@ -14413,19 +14174,13 @@ class ReticulumMeshChat:
deadline = time.time() + self._lxmf_path_wait_seconds()
if not RNS.Transport.has_path(destination_hash_bytes):
RNS.Transport.request_path(destination_hash_bytes)
- while (
- not RNS.Transport.has_path(destination_hash_bytes)
- and time.time() < deadline
- ):
+ while not RNS.Transport.has_path(destination_hash_bytes) and time.time() < deadline:
await asyncio.sleep(0.1)
if RNS.Transport.has_path(destination_hash_bytes):
return True
RNS.Transport.request_path(destination_hash_bytes)
deadline = time.time() + max(15.0, self._lxmf_path_wait_seconds() * 0.5)
- while (
- not RNS.Transport.has_path(destination_hash_bytes)
- and time.time() < deadline
- ):
+ while not RNS.Transport.has_path(destination_hash_bytes) and time.time() < deadline:
await asyncio.sleep(0.1)
return RNS.Transport.has_path(destination_hash_bytes)
@@ -14438,14 +14193,14 @@ class ReticulumMeshChat:
image_field: LxmfImageField = None,
audio_field: LxmfAudioField = None,
file_attachments_field: LxmfFileAttachmentsField = None,
- telemetry_data: bytes = None,
- commands: list = None,
- delivery_method: str = None,
+ telemetry_data: bytes | None = None,
+ commands: list | None = None,
+ delivery_method: str | None = None,
title: str = "",
- sender_identity_hash: str = None,
- reply_to_hash: str = None,
- reply_quoted_content: str = None,
- app_extensions: dict = None,
+ sender_identity_hash: str | None = None,
+ reply_to_hash: str | None = None,
+ reply_quoted_content: str | None = None,
+ app_extensions: dict | None = None,
no_display: bool = False,
context=None,
) -> LXMF.LXMessage:
@@ -14523,8 +14278,7 @@ class ReticulumMeshChat:
if sender_identity_hash is not None:
if (
ctx.forwarding_manager
- and sender_identity_hash
- in ctx.forwarding_manager.forwarding_destinations
+ and sender_identity_hash in ctx.forwarding_manager.forwarding_destinations
):
source_destination = ctx.forwarding_manager.forwarding_destinations[
sender_identity_hash
@@ -15332,7 +15086,6 @@ class ReticulumMeshChat:
# reads the lxmf display name from the provided base64 app data
# returns true if the conversation has messages newer than the last read at timestamp
- @staticmethod
def is_lxmf_conversation_unread(self, destination_hash):
return self.database.messages.is_conversation_unread(destination_hash)
diff --git a/meshchatx/src/backend/async_utils.py b/meshchatx/src/backend/async_utils.py
index e19cccfe..86364b32 100644
--- a/meshchatx/src/backend/async_utils.py
+++ b/meshchatx/src/backend/async_utils.py
@@ -4,12 +4,13 @@ import asyncio
import sys
import threading
from collections.abc import Coroutine
+from typing import Any, ClassVar
class AsyncUtils:
main_loop: asyncio.AbstractEventLoop | None = None
- _pending_futures: list = []
- _pending_coroutines: list = []
+ _pending_futures: ClassVar[list[Any]] = []
+ _pending_coroutines: ClassVar[list[Any]] = []
_futures_lock = threading.Lock()
_FUTURES_SWEEP_THRESHOLD = 64
@@ -70,10 +71,7 @@ class AsyncUtils:
)
with AsyncUtils._futures_lock:
AsyncUtils._pending_futures.append(future)
- if (
- len(AsyncUtils._pending_futures)
- >= AsyncUtils._FUTURES_SWEEP_THRESHOLD
- ):
+ if len(AsyncUtils._pending_futures) >= AsyncUtils._FUTURES_SWEEP_THRESHOLD:
AsyncUtils._pending_futures = [
f for f in AsyncUtils._pending_futures if not f.done()
]
diff --git a/meshchatx/src/backend/audio_codec.py b/meshchatx/src/backend/audio_codec.py
index d09167f5..ec65a7ae 100644
--- a/meshchatx/src/backend/audio_codec.py
+++ b/meshchatx/src/backend/audio_codec.py
@@ -79,13 +79,9 @@ def _decode_with_wave(data: bytes):
if sample_width == 2:
samples = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
elif sample_width == 1:
- samples = (
- np.frombuffer(raw, dtype=np.uint8).astype(np.float32) - 128.0
- ) / 128.0
+ samples = (np.frombuffer(raw, dtype=np.uint8).astype(np.float32) - 128.0) / 128.0
elif sample_width == 4:
- samples = (
- np.frombuffer(raw, dtype=np.int32).astype(np.float32) / 2147483648.0
- )
+ samples = np.frombuffer(raw, dtype=np.int32).astype(np.float32) / 2147483648.0
else:
return None
@@ -113,9 +109,7 @@ def _decode_with_miniaudio(data: bytes):
return None
if decoded.num_frames <= 0 or decoded.nchannels <= 0:
return None
- samples = (
- np.frombuffer(decoded.samples, dtype=np.int16).astype(np.float32) / 32768.0
- )
+ samples = np.frombuffer(decoded.samples, dtype=np.int16).astype(np.float32) / 32768.0
return DecodedAudio(
samples=samples.reshape(-1, decoded.nchannels),
samplerate=decoded.sample_rate,
@@ -128,7 +122,6 @@ def _decode_with_lxst_opus(data: bytes):
return None
try:
import numpy as np
-
from LXST.Codecs.libs.pyogg import OpusFile
except ImportError:
return None
@@ -258,7 +251,6 @@ def encode_pcm_to_ogg_opus(
frame loss and no trailing silence padding.
"""
import numpy as np
-
from LXST.Codecs import Opus
from LXST.Codecs.libs.pyogg import OggOpusWriter, OpusBufferedEncoder
diff --git a/meshchatx/src/backend/auto_propagation_manager.py b/meshchatx/src/backend/auto_propagation_manager.py
index df53ef66..b6d0b761 100644
--- a/meshchatx/src/backend/auto_propagation_manager.py
+++ b/meshchatx/src/backend/auto_propagation_manager.py
@@ -120,7 +120,7 @@ class AutoPropagationManager:
continue
try:
dest_hash = bytes.fromhex(dest_hex)
- except Exception:
+ except ValueError:
continue
if RNS.Transport.has_path(dest_hash):
hops = RNS.Transport.hops_to(dest_hash)
@@ -134,9 +134,7 @@ class AutoPropagationManager:
if not sorted_candidates:
return
- previous_hex = (
- self.config.lxmf_preferred_propagation_node_destination_hash.get()
- )
+ previous_hex = self.config.lxmf_preferred_propagation_node_destination_hash.get()
ordered: list[tuple[int, str]] = []
seen_hex: set[str] = set()
if previous_hex and previous_hex in best_by_hex:
@@ -150,7 +148,7 @@ class AutoPropagationManager:
for _hops, node_hex in ordered:
try:
dest_hash = bytes.fromhex(node_hex)
- except Exception:
+ except ValueError:
continue
if not await self._wait_for_path(dest_hash, PATH_WAIT_SECONDS):
diff --git a/meshchatx/src/backend/bot_handler.py b/meshchatx/src/backend/bot_handler.py
index 58cad14d..984cdfc6 100644
--- a/meshchatx/src/backend/bot_handler.py
+++ b/meshchatx/src/backend/bot_handler.py
@@ -46,14 +46,11 @@ class BotHandler:
for entry in self.bots_state:
if "storage_dir" in entry:
entry["storage_dir"] = os.path.abspath(entry["storage_dir"])
- if "bot_config_dir" in entry and entry["bot_config_dir"]:
+ if entry.get("bot_config_dir"):
entry["bot_config_dir"] = os.path.abspath(
os.path.expanduser(entry["bot_config_dir"]),
)
- if (
- "reticulum_config_dir" in entry
- and entry["reticulum_config_dir"]
- ):
+ if entry.get("reticulum_config_dir"):
entry["reticulum_config_dir"] = os.path.abspath(
os.path.expanduser(entry["reticulum_config_dir"]),
)
@@ -150,16 +147,10 @@ class BotHandler:
# Try running instance first
instance = self.running_bots.get(bot_id, {}).get("instance")
- if (
- instance
- and getattr(instance, "bot", None)
- and getattr(instance.bot, "local", None)
- ):
+ if instance and getattr(instance, "bot", None) and getattr(instance.bot, "local", None):
with contextlib.suppress(Exception):
lh = instance.bot.local.hash
- address_full = (
- lh.hex() if isinstance(lh, (bytes, bytearray)) else None
- )
+ address_full = lh.hex() if isinstance(lh, (bytes, bytearray)) else None
if address_full:
address_full = self._normalize_lxmf_hash_hex(address_full)
if address_full:
@@ -173,9 +164,7 @@ class BotHandler:
destination = RNS.Destination(identity, "lxmf", "delivery")
address_full = self._normalize_lxmf_hash_hex(destination.hash)
if address_full:
- address_pretty = RNS.prettyhexrep(
- bytes.fromhex(address_full)
- )
+ address_pretty = RNS.prettyhexrep(bytes.fromhex(address_full))
if address_full is None:
address_full = self._read_lxmf_address_sidecar(entry.get("storage_dir"))
@@ -255,7 +244,7 @@ class BotHandler:
entry["reticulum_config_dir"],
]
- proc = subprocess.Popen(cmd, cwd=bot_storage_dir) # noqa: S603
+ proc = subprocess.Popen(cmd, cwd=bot_storage_dir)
entry["pid"] = proc.pid
self._save_state()
@@ -284,7 +273,7 @@ class BotHandler:
if sys.platform.startswith("win"):
# Use absolute path if possible to avoid S607
taskkill = shutil.which("taskkill") or "taskkill"
- subprocess.run( # noqa: S603
+ subprocess.run(
[taskkill, "/PID", str(pid), "/T", "/F"],
check=False,
timeout=5,
diff --git a/meshchatx/src/backend/bot_process.py b/meshchatx/src/backend/bot_process.py
index fd3fc650..92052899 100644
--- a/meshchatx/src/backend/bot_process.py
+++ b/meshchatx/src/backend/bot_process.py
@@ -60,9 +60,7 @@ def main():
else:
config_path = os.path.join(os.path.abspath(args.storage), "config")
os.makedirs(config_path, exist_ok=True)
- reticulum_config_dir = os.path.abspath(
- os.path.expanduser(args.reticulum_config_dir)
- )
+ reticulum_config_dir = os.path.abspath(os.path.expanduser(args.reticulum_config_dir))
os.makedirs(reticulum_config_dir, exist_ok=True)
BotCls = TEMPLATE_MAP[args.template]
diff --git a/meshchatx/src/backend/bot_templates.py b/meshchatx/src/backend/bot_templates.py
index c9268c39..7edac8d2 100644
--- a/meshchatx/src/backend/bot_templates.py
+++ b/meshchatx/src/backend/bot_templates.py
@@ -2,7 +2,7 @@
import re
import time
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
from lxmfy import IconAppearance, LXMFBot, pack_icon_appearance_field
@@ -141,7 +141,7 @@ class NoteBotTemplate(StoppableBot):
note = {
"text": " ".join(ctx.args),
- "timestamp": datetime.now().isoformat(),
+ "timestamp": datetime.now(UTC).isoformat(),
"tags": [w[1:] for w in ctx.args if w.startswith("#")],
}
@@ -162,23 +162,17 @@ class NoteBotTemplate(StoppableBot):
if not ctx.args:
response = "Your Notes:\n"
for i, note in enumerate(notes[-10:], 1):
- tags = (
- " ".join(f"#{tag}" for tag in note["tags"])
- if note["tags"]
- else ""
- )
+ tags = " ".join(f"#{tag}" for tag in note["tags"]) if note["tags"] else ""
response += f"{i}. {note['text']} {tags}\n"
if len(notes) > 10:
- response += f"\nShowing last 10 of {len(notes)} notes. Use /notes all to see all."
+ response += (
+ f"\nShowing last 10 of {len(notes)} notes. Use /notes all to see all."
+ )
ctx.reply(response)
elif ctx.args[0] == "all":
response = "All Your Notes:\n"
for i, note in enumerate(notes, 1):
- tags = (
- " ".join(f"#{tag}" for tag in note["tags"])
- if note["tags"]
- else ""
- )
+ tags = " ".join(f"#{tag}" for tag in note["tags"]) if note["tags"] else ""
response += f"{i}. {note['text']} {tags}\n"
ctx.reply(response)
@@ -252,7 +246,7 @@ class ReminderBotTemplate(StoppableBot):
ctx.reply("Invalid time format. Use combinations of d, h, m")
return
- remind_time = datetime.now() + timedelta(minutes=total_minutes)
+ remind_time = datetime.now(UTC) + timedelta(minutes=total_minutes)
reminder = {
"user": ctx.sender,
"message": message,
diff --git a/meshchatx/src/backend/config_manager.py b/meshchatx/src/backend/config_manager.py
index ba9a310c..211cbc2c 100644
--- a/meshchatx/src/backend/config_manager.py
+++ b/meshchatx/src/backend/config_manager.py
@@ -66,12 +66,10 @@ class ConfigManager:
"lxmf_preferred_propagation_node_auto_select",
False,
)
- self.lxmf_preferred_propagation_node_auto_sync_interval_seconds = (
- self.IntConfig(
- self,
- "lxmf_preferred_propagation_node_auto_sync_interval_seconds",
- 0,
- )
+ self.lxmf_preferred_propagation_node_auto_sync_interval_seconds = self.IntConfig(
+ self,
+ "lxmf_preferred_propagation_node_auto_sync_interval_seconds",
+ 0,
)
self.lxmf_preferred_propagation_node_last_synced_at = self.IntConfig(
self,
@@ -449,7 +447,7 @@ class ConfigManager:
self.key = key
self.default_value = default_value
- def get(self, default_value: str = None) -> str | None:
+ def get(self, default_value: str | None = None) -> str | None:
_default_value = default_value or self.default_value
return self.manager.get(self.key, default_value=_default_value)
diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py
index a2778f07..afc8640e 100644
--- a/meshchatx/src/backend/database/__init__.py
+++ b/meshchatx/src/backend/database/__init__.py
@@ -125,8 +125,7 @@ class Database:
return None
try:
with open(path) as f:
- data = json.load(f)
- return data
+ return json.load(f)
except (OSError, json.JSONDecodeError):
return None
@@ -181,7 +180,7 @@ class Database:
_log.warning("DB open health check: no result")
else:
first = integrity_rows[0]
- val = list(first.values())[0] if isinstance(first, dict) else first[0]
+ val = next(iter(first.values())) if isinstance(first, dict) else first[0]
if val != "ok":
issues.append(f"Database integrity check failed: {val!s}")
_log.warning("DB open health check: %s", val)
@@ -228,7 +227,7 @@ class Database:
_log.warning("DB close health check: no result")
else:
first = integrity_rows[0]
- val = list(first.values())[0] if isinstance(first, dict) else first[0]
+ val = next(iter(first.values())) if isinstance(first, dict) else first[0]
if val != "ok":
issues.append(f"Database integrity check failed: {val!s}")
_log.warning("DB close health check: integrity failed")
@@ -266,9 +265,7 @@ class Database:
page_size = self._get_pragma_value("page_size", 0) or 0
page_count = self._get_pragma_value("page_count", 0) or 0
freelist_pages = self._get_pragma_value("freelist_count", 0) or 0
- freelist_bytes = (
- page_size * freelist_pages if page_size > 0 and freelist_pages > 0 else 0
- )
+ freelist_bytes = page_size * freelist_pages if page_size > 0 and freelist_pages > 0 else 0
if freelist_bytes > 0:
free_bytes = freelist_bytes
else:
@@ -312,7 +309,7 @@ class Database:
}
except Exception as e:
# Wrap in a cleaner error message
- raise Exception(f"Database vacuum failed: {e!s}")
+ raise Exception(f"Database vacuum failed: {e!s}") from e
def run_database_recovery(self):
actions = []
diff --git a/meshchatx/src/backend/database/announces.py b/meshchatx/src/backend/database/announces.py
index 125f59d8..4235a79a 100644
--- a/meshchatx/src/backend/database/announces.py
+++ b/meshchatx/src/backend/database/announces.py
@@ -39,7 +39,7 @@ class AnnounceDAO:
update_set = ", ".join(update_parts)
query = (
- f"INSERT INTO announces ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) " # noqa: S608
+ f"INSERT INTO announces ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) "
f"ON CONFLICT(destination_hash) DO UPDATE SET {update_set}, updated_at = EXCLUDED.updated_at"
)
diff --git a/meshchatx/src/backend/database/contacts.py b/meshchatx/src/backend/database/contacts.py
index db7cd0cd..171b1383 100644
--- a/meshchatx/src/backend/database/contacts.py
+++ b/meshchatx/src/backend/database/contacts.py
@@ -125,7 +125,7 @@ class ContactsDAO:
return
updates.append("updated_at = CURRENT_TIMESTAMP")
- query = f"UPDATE contacts SET {', '.join(updates)} WHERE id = ?" # noqa: S608
+ query = f"UPDATE contacts SET {', '.join(updates)} WHERE id = ?"
params.append(contact_id)
self.provider.execute(query, tuple(params))
diff --git a/meshchatx/src/backend/database/gifs.py b/meshchatx/src/backend/database/gifs.py
index c7b36a57..e3968199 100644
--- a/meshchatx/src/backend/database/gifs.py
+++ b/meshchatx/src/backend/database/gifs.py
@@ -192,8 +192,7 @@ class UserGifsDAO:
b64 = item.get("image_bytes_b64")
src = item.get("source_message_hash")
usage = int(item.get("usage_count") or 0)
- if usage < 0:
- usage = 0
+ usage = max(usage, 0)
try:
raw = base64.b64decode(b64, validate=False)
except (ValueError, TypeError):
@@ -220,10 +219,7 @@ class UserGifsDAO:
(identity_hash, ch),
)
- if (
- self.count_for_identity(identity_hash)
- >= gif_utils.MAX_GIFS_PER_IDENTITY
- ):
+ if self.count_for_identity(identity_hash) >= gif_utils.MAX_GIFS_PER_IDENTITY:
errors.append("gif_limit_reached")
break
diff --git a/meshchatx/src/backend/database/legacy_migrator.py b/meshchatx/src/backend/database/legacy_migrator.py
index bec7f723..3c9d3b27 100644
--- a/meshchatx/src/backend/database/legacy_migrator.py
+++ b/meshchatx/src/backend/database/legacy_migrator.py
@@ -60,7 +60,7 @@ class LegacyMigrator:
if res and res["count"] > 0:
# Already have data, don't auto-migrate
return False
- except Exception: # noqa: S110
+ except Exception:
# Table doesn't exist yet, which is fine
# We use a broad Exception here as the database might not even be initialized
pass
@@ -104,7 +104,9 @@ class LegacyMigrator:
try:
# Check if table exists in legacy DB
# We use a f-string here for the alias and table name, which are controlled by us
- check_query = f"SELECT name FROM {alias}.sqlite_master WHERE type='table' AND name=?" # noqa: S608
+ check_query = (
+ f"SELECT name FROM {alias}.sqlite_master WHERE type='table' AND name=?"
+ )
res = self.provider.fetchone(check_query, (table,))
if res:
@@ -137,7 +139,7 @@ class LegacyMigrator:
cols_str = ", ".join(common_columns)
# We use INSERT OR IGNORE to avoid duplicates
# The table and columns are controlled by us
- migrate_query = f"INSERT OR IGNORE INTO {table} ({cols_str}) SELECT {cols_str} FROM {alias}.{table}" # noqa: S608
+ migrate_query = f"INSERT OR IGNORE INTO {table} ({cols_str}) SELECT {cols_str} FROM {alias}.{table}"
self.provider.execute(migrate_query)
print(
f" - Migrated table: {table} ({len(common_columns)} columns)",
diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py
index 6ba11754..69736b83 100644
--- a/meshchatx/src/backend/database/messages.py
+++ b/meshchatx/src/backend/database/messages.py
@@ -44,7 +44,7 @@ class MessageDAO:
update_set = ", ".join([f"{f} = EXCLUDED.{f}" for f in fields if f != "hash"])
query = (
- f"INSERT INTO lxmf_messages ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) " # noqa: S608
+ f"INSERT INTO lxmf_messages ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) "
f"ON CONFLICT(hash) DO UPDATE SET {update_set}, updated_at = EXCLUDED.updated_at"
)
@@ -150,7 +150,7 @@ class MessageDAO:
return
placeholders = ", ".join(["?"] * len(message_hashes))
self.provider.execute(
- f"DELETE FROM lxmf_messages WHERE hash IN ({placeholders})", # noqa: S608
+ f"DELETE FROM lxmf_messages WHERE hash IN ({placeholders})",
tuple(message_hashes),
)
@@ -234,7 +234,7 @@ class MessageDAO:
) m2 ON m1.peer_hash = m2.peer_hash AND m1.timestamp = m2.max_ts
GROUP BY m1.peer_hash
ORDER BY m1.timestamp DESC
- """ # noqa: S608
+ """
return self.provider.fetchall(query)
def mark_conversation_as_read(self, destination_hash):
@@ -325,7 +325,7 @@ class MessageDAO:
LEFT JOIN lxmf_conversation_read_state r ON r.destination_hash = m.peer_hash
WHERE m.peer_hash IN ({placeholders}) AND m.is_incoming = 1
GROUP BY m.peer_hash
- """ # noqa: S608
+ """
rows = self.provider.fetchall(query, destination_hashes)
unread_states = {}
@@ -351,7 +351,7 @@ class MessageDAO:
return {}
placeholders = ", ".join(["?"] * len(destination_hashes))
rows = self.provider.fetchall(
- f"SELECT peer_hash, COUNT(*) as count FROM lxmf_messages WHERE state = 'failed' AND peer_hash IN ({placeholders}) GROUP BY peer_hash", # noqa: S608
+ f"SELECT peer_hash, COUNT(*) as count FROM lxmf_messages WHERE state = 'failed' AND peer_hash IN ({placeholders}) GROUP BY peer_hash",
tuple(destination_hashes),
)
return {row["peer_hash"]: row["count"] for row in rows}
@@ -367,7 +367,7 @@ class MessageDAO:
WHERE peer_hash IN ({placeholders})
AND fields IS NOT NULL AND fields != '{{}}' AND fields != ''
GROUP BY peer_hash
- """ # noqa: S608
+ """
rows = self.provider.fetchall(query, destination_hashes)
return {row["peer_hash"]: True for row in rows}
@@ -405,7 +405,7 @@ class MessageDAO:
]
columns = ", ".join(fields)
placeholders = ", ".join(["?"] * len(fields))
- query = f"INSERT INTO lxmf_forwarding_mappings ({columns}, created_at) VALUES ({placeholders}, ?)" # noqa: S608
+ query = f"INSERT INTO lxmf_forwarding_mappings ({columns}, created_at) VALUES ({placeholders}, ?)"
params = [data.get(f) for f in fields]
params.append(datetime.now(UTC).isoformat())
self.provider.execute(query, params)
@@ -523,7 +523,7 @@ class MessageDAO:
if folder_id is None:
placeholders = ", ".join(["?"] * len(peer_hashes))
self.provider.execute(
- f"DELETE FROM lxmf_conversation_folders WHERE peer_hash IN ({placeholders})", # noqa: S608
+ f"DELETE FROM lxmf_conversation_folders WHERE peer_hash IN ({placeholders})",
tuple(peer_hashes),
)
else:
diff --git a/meshchatx/src/backend/database/misc.py b/meshchatx/src/backend/database/misc.py
index 84a007ee..94dc94cf 100644
--- a/meshchatx/src/backend/database/misc.py
+++ b/meshchatx/src/backend/database/misc.py
@@ -97,7 +97,7 @@ class MiscDAO:
return []
placeholders = ", ".join(["?"] * len(destination_hashes))
return self.provider.fetchall(
- f"SELECT * FROM lxmf_user_icons WHERE destination_hash IN ({placeholders})", # noqa: S608
+ f"SELECT * FROM lxmf_user_icons WHERE destination_hash IN ({placeholders})",
tuple(destination_hashes),
)
@@ -177,9 +177,7 @@ class MiscDAO:
params.append(destination_hash)
if query:
like_term = f"%{query}%"
- sql += (
- " AND (destination_hash LIKE ? OR page_path LIKE ? OR content LIKE ?)"
- )
+ sql += " AND (destination_hash LIKE ? OR page_path LIKE ? OR content LIKE ?)"
params.extend([like_term, like_term, like_term])
sql += " ORDER BY created_at DESC"
@@ -189,7 +187,7 @@ class MiscDAO:
if ids:
placeholders = ", ".join(["?"] * len(ids))
self.provider.execute(
- f"DELETE FROM archived_pages WHERE id IN ({placeholders})", # noqa: S608
+ f"DELETE FROM archived_pages WHERE id IN ({placeholders})",
tuple(ids),
)
elif destination_hash and page_path:
@@ -244,7 +242,7 @@ class MiscDAO:
set_clause = ", ".join([f"{k} = ?" for k in filtered_kwargs])
params = list(filtered_kwargs.values())
params.append(task_id)
- query = f"UPDATE crawl_tasks SET {set_clause} WHERE id = ?" # noqa: S608
+ query = f"UPDATE crawl_tasks SET {set_clause} WHERE id = ?"
self.provider.execute(query, params)
def get_pending_or_failed_crawl_tasks(self, max_retries, max_concurrent):
@@ -281,7 +279,7 @@ class MiscDAO:
if notification_ids:
placeholders = ", ".join(["?"] * len(notification_ids))
self.provider.execute(
- f"UPDATE notifications SET is_viewed = 1 WHERE id IN ({placeholders})", # noqa: S608
+ f"UPDATE notifications SET is_viewed = 1 WHERE id IN ({placeholders})",
notification_ids,
)
else:
diff --git a/meshchatx/src/backend/database/provider.py b/meshchatx/src/backend/database/provider.py
index b4b50698..30d021cb 100644
--- a/meshchatx/src/backend/database/provider.py
+++ b/meshchatx/src/backend/database/provider.py
@@ -74,13 +74,10 @@ class DatabaseProvider:
if isinstance(params, dict):
params = {
- k: (v.isoformat() if isinstance(v, datetime) else v)
- for k, v in params.items()
+ k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in params.items()
}
else:
- params = tuple(
- (p.isoformat() if isinstance(p, datetime) else p) for p in params
- )
+ params = tuple((p.isoformat() if isinstance(p, datetime) else p) for p in params)
if params:
cursor.execute(query, params)
@@ -148,7 +145,7 @@ class DatabaseProvider:
try:
self._memory_connection.commit()
self._memory_connection.close()
- except Exception: # noqa: S110
+ except Exception:
pass
self._memory_connection = None
@@ -156,7 +153,7 @@ class DatabaseProvider:
try:
self.commit() # Ensure everything is saved
self._local.connection.close()
- except Exception: # noqa: S110
+ except Exception:
pass
del self._local.connection
@@ -166,7 +163,7 @@ class DatabaseProvider:
try:
self._memory_connection.commit()
self._memory_connection.close()
- except Exception: # noqa: S110
+ except Exception:
pass
self._memory_connection = None
@@ -175,7 +172,7 @@ class DatabaseProvider:
try:
loc.connection.commit()
loc.connection.close()
- except Exception: # noqa: S110
+ except Exception:
pass
del loc.connection
diff --git a/meshchatx/src/backend/database/ringtones.py b/meshchatx/src/backend/database/ringtones.py
index aa540ae8..0395a19b 100644
--- a/meshchatx/src/backend/database/ringtones.py
+++ b/meshchatx/src/backend/database/ringtones.py
@@ -29,9 +29,7 @@ class RingtoneDAO:
display_name = filename
# check if this is the first ringtone, if so make it primary
- count = self.provider.fetchone("SELECT COUNT(*) as count FROM ringtones")[
- "count"
- ]
+ count = self.provider.fetchone("SELECT COUNT(*) as count FROM ringtones")["count"]
is_primary = 1 if count == 0 else 0
cursor = self.provider.execute(
diff --git a/meshchatx/src/backend/database/sticker_packs.py b/meshchatx/src/backend/database/sticker_packs.py
index 8c153f81..20843030 100644
--- a/meshchatx/src/backend/database/sticker_packs.py
+++ b/meshchatx/src/backend/database/sticker_packs.py
@@ -15,7 +15,6 @@ import time
from meshchatx.src.backend import sticker_pack_utils, sticker_utils
-
_PACK_COLUMNS = (
"id, identity_hash, title, short_name, description, pack_type, author, "
"is_strict, cover_sticker_id, sort_order, created_at, updated_at"
@@ -82,10 +81,7 @@ class UserStickerPacksDAO:
is_strict: bool = True,
) -> dict:
"""Create a new pack. Raises ``ValueError`` on quota or short_name clash."""
- if (
- self.count_for_identity(identity_hash)
- >= sticker_utils.MAX_STICKER_PACKS_PER_IDENTITY
- ):
+ if self.count_for_identity(identity_hash) >= sticker_utils.MAX_STICKER_PACKS_PER_IDENTITY:
msg = "pack_limit_reached"
raise ValueError(msg)
sn = sticker_pack_utils.sanitize_pack_short_name(short_name)
@@ -154,11 +150,7 @@ class UserStickerPacksDAO:
if pack_type is not None
else existing["pack_type"]
)
- new_cover = (
- existing["cover_sticker_id"]
- if cover_sticker_id is ...
- else cover_sticker_id
- )
+ new_cover = existing["cover_sticker_id"] if cover_sticker_id is ... else cover_sticker_id
cur = self.provider.execute(
"""
UPDATE user_sticker_packs
diff --git a/meshchatx/src/backend/database/stickers.py b/meshchatx/src/backend/database/stickers.py
index 2a167c73..c5cdfb1f 100644
--- a/meshchatx/src/backend/database/stickers.py
+++ b/meshchatx/src/backend/database/stickers.py
@@ -6,7 +6,6 @@ import time
from meshchatx.src.backend import sticker_utils
-
_STICKER_SUMMARY_COLUMNS = (
"id, identity_hash, name, image_type, length(image_blob) AS image_size, "
"content_hash, source_message_hash, pack_id, emoji, width, height, "
@@ -188,10 +187,7 @@ class UserStickersDAO:
metadata so the picker can render the sticker correctly without
re-parsing.
"""
- if (
- self.count_for_identity(identity_hash)
- >= sticker_utils.MAX_STICKERS_PER_IDENTITY
- ):
+ if self.count_for_identity(identity_hash) >= sticker_utils.MAX_STICKERS_PER_IDENTITY:
msg = "sticker_limit_reached"
raise ValueError(msg)
@@ -349,10 +345,7 @@ class UserStickersDAO:
(identity_hash, ch),
)
- if (
- self.count_for_identity(identity_hash)
- >= sticker_utils.MAX_STICKERS_PER_IDENTITY
- ):
+ if self.count_for_identity(identity_hash) >= sticker_utils.MAX_STICKERS_PER_IDENTITY:
errors.append("sticker_limit_reached")
break
diff --git a/meshchatx/src/backend/docs_manager.py b/meshchatx/src/backend/docs_manager.py
index e943a5c7..6fe9c9d9 100644
--- a/meshchatx/src/backend/docs_manager.py
+++ b/meshchatx/src/backend/docs_manager.py
@@ -206,7 +206,7 @@ class DocsManager:
try:
for file in os.listdir(src_docs):
- if file.endswith(".md") or file.endswith(".txt"):
+ if file.endswith((".md", ".txt")):
src_path = os.path.join(src_docs, file)
dest_path = os.path.join(self.meshchatx_docs_dir, file)
@@ -264,9 +264,7 @@ class DocsManager:
def has_meshchatx_docs(self):
return (
- any(
- f.endswith((".md", ".txt")) for f in os.listdir(self.meshchatx_docs_dir)
- )
+ any(f.endswith((".md", ".txt")) for f in os.listdir(self.meshchatx_docs_dir))
if os.path.exists(self.meshchatx_docs_dir)
else False
)
@@ -635,7 +633,7 @@ class DocsManager:
if not os.path.isdir(path):
return
try:
- os.chmod(path, 0o755)
+ os.chmod(path, 0o755) # noqa: S103
except OSError:
pass
diff --git a/meshchatx/src/backend/forwarding_manager.py b/meshchatx/src/backend/forwarding_manager.py
index 0577363c..963d06c0 100644
--- a/meshchatx/src/backend/forwarding_manager.py
+++ b/meshchatx/src/backend/forwarding_manager.py
@@ -144,10 +144,7 @@ class ForwardingManager:
for link in list(RNS.Transport.active_links):
match = False
if hasattr(link, "destination") and link.destination:
- if (
- hasattr(link.destination, "identity")
- and link.destination.identity
- ):
+ if hasattr(link.destination, "identity") and link.destination.identity:
if link.destination.identity.hash == ih:
match = True
if match:
diff --git a/meshchatx/src/backend/gif_utils.py b/meshchatx/src/backend/gif_utils.py
index 5b894b80..34c65c47 100644
--- a/meshchatx/src/backend/gif_utils.py
+++ b/meshchatx/src/backend/gif_utils.py
@@ -143,8 +143,7 @@ def validate_export_document(data: object) -> list[dict]:
usage_int = int(usage) if usage is not None else 0
except (TypeError, ValueError):
usage_int = 0
- if usage_int < 0:
- usage_int = 0
+ usage_int = max(usage_int, 0)
out.append(
{
"name": name if isinstance(name, str) else None,
diff --git a/meshchatx/src/backend/identity_context.py b/meshchatx/src/backend/identity_context.py
index 9e342241..62289015 100644
--- a/meshchatx/src/backend/identity_context.py
+++ b/meshchatx/src/backend/identity_context.py
@@ -158,10 +158,7 @@ class IdentityContext:
self.config = ConfigManager(self.database)
# Apply overrides from CLI/ENV if provided
- if (
- hasattr(self.app, "gitea_base_url_override")
- and self.app.gitea_base_url_override
- ):
+ if hasattr(self.app, "gitea_base_url_override") and self.app.gitea_base_url_override:
self.config.gitea_base_url.set(self.app.gitea_base_url_override)
self.message_handler = MessageHandler(self.database)
@@ -238,9 +235,7 @@ class IdentityContext:
# Restore preferred propagation node on startup
with contextlib.suppress(Exception):
- preferred_node = (
- self.config.lxmf_preferred_propagation_node_destination_hash.get()
- )
+ preferred_node = self.config.lxmf_preferred_propagation_node_destination_hash.get()
if preferred_node:
self.app.set_active_propagation_node(preferred_node, context=self)
@@ -289,9 +284,7 @@ class IdentityContext:
storage_dir=self.storage_path,
db=self.database,
)
- self.telephone_manager.get_name_for_identity_hash = (
- self.app.get_name_for_identity_hash
- )
+ self.telephone_manager.get_name_for_identity_hash = self.app.get_name_for_identity_hash
self.telephone_manager.on_initiation_status_callback = lambda status, target: (
self.app.on_telephone_initiation_status(
status,
@@ -319,9 +312,7 @@ class IdentityContext:
telephone_manager=self.telephone_manager,
storage_dir=self.storage_path,
)
- self.voicemail_manager.get_name_for_identity_hash = (
- self.app.get_name_for_identity_hash
- )
+ self.voicemail_manager.get_name_for_identity_hash = self.app.get_name_for_identity_hash
self.voicemail_manager.on_new_voicemail_callback = lambda vm: (
self.app.on_new_voicemail_received(vm, context=self)
)
@@ -365,9 +356,7 @@ class IdentityContext:
# start background thread for auto syncing propagation nodes
thread = threading.Thread(
target=asyncio.run,
- args=(
- self.app.announce_sync_propagation_nodes(self.session_id, context=self),
- ),
+ args=(self.app.announce_sync_propagation_nodes(self.session_id, context=self),),
)
thread.daemon = True
thread.start()
@@ -430,28 +419,24 @@ class IdentityContext:
),
AnnounceHandler(
"lxmf.propagation",
- lambda aspect, dh, ai, ad, aph: (
- self.app.on_lxmf_propagation_announce_received(
- aspect,
- dh,
- ai,
- ad,
- aph,
- context=self,
- )
+ lambda aspect, dh, ai, ad, aph: self.app.on_lxmf_propagation_announce_received(
+ aspect,
+ dh,
+ ai,
+ ad,
+ aph,
+ context=self,
),
),
AnnounceHandler(
"nomadnetwork.node",
- lambda aspect, dh, ai, ad, aph: (
- self.app.on_nomadnet_node_announce_received(
- aspect,
- dh,
- ai,
- ad,
- aph,
- context=self,
- )
+ lambda aspect, dh, ai, ad, aph: self.app.on_nomadnet_node_announce_received(
+ aspect,
+ dh,
+ ai,
+ ad,
+ aph,
+ context=self,
),
),
]
diff --git a/meshchatx/src/backend/identity_manager.py b/meshchatx/src/backend/identity_manager.py
index 816807e8..617dac3f 100644
--- a/meshchatx/src/backend/identity_manager.py
+++ b/meshchatx/src/backend/identity_manager.py
@@ -53,7 +53,7 @@ class IdentityManager:
try:
with open(identity_file, "rb") as f:
result[identity_hash] = f.read()
- except Exception:
+ except OSError:
continue
return result
@@ -148,8 +148,7 @@ class IdentityManager:
"lxmf_address": lxmf_address,
"lxst_address": lxst_address,
"is_current": (
- current_identity_hash is not None
- and identity_hash == current_identity_hash
+ current_identity_hash is not None and identity_hash == current_identity_hash
),
},
)
@@ -234,7 +233,7 @@ class IdentityManager:
return self._save_new_identity(identity, "Restored Identity")
except Exception as exc:
- raise ValueError(f"Failed to restore identity: {exc}")
+ raise ValueError(f"Failed to restore identity: {exc}") from exc
def restore_identity_from_base32(self, base32_value: str) -> dict:
try:
diff --git a/meshchatx/src/backend/integrity_manager.py b/meshchatx/src/backend/integrity_manager.py
index f524acc1..b032d41f 100644
--- a/meshchatx/src/backend/integrity_manager.py
+++ b/meshchatx/src/backend/integrity_manager.py
@@ -8,6 +8,7 @@ import os
import sqlite3
from datetime import UTC, datetime
from pathlib import Path
+from typing import ClassVar
class IntegrityManager:
@@ -15,7 +16,7 @@ class IntegrityManager:
# Files and directories that are frequently modified by RNS/LXMF or SQLite
# and should be ignored during integrity checks.
- IGNORED_PATTERNS = [
+ IGNORED_PATTERNS: ClassVar[list[str]] = [
"*-wal",
"*-shm",
"*-journal",
@@ -46,10 +47,7 @@ class IntegrityManager:
# to avoid accidentally ignoring important files with similar names.
if "lxmf_router" in path_parts:
# Added more volatile LXMF patterns
- if any(
- part in ["announces", "storage", "identities", "tmp"]
- for part in path_parts
- ):
+ if any(part in ["announces", "storage", "identities", "tmp"] for part in path_parts):
return True
# Specifically ignore stamp costs which are frequently updated
@@ -135,7 +133,7 @@ class IntegrityManager:
m_id = manifest.get("identity", "Unknown")
# Always check for identity mismatch first as it's a fundamental security issue
- if self.identity_hash and m_id != "Unknown" and self.identity_hash != m_id:
+ if self.identity_hash and m_id not in ("Unknown", self.identity_hash):
issues.append(f"Identity mismatch! Manifest belongs to: {m_id}")
# Check Database (Math-based structural check + Entropy stability + Hash)
@@ -156,10 +154,7 @@ class IntegrityManager:
actual_entropy = self._calculate_entropy(self.database_path)
saved_entropy = manifest_metadata.get(db_rel, {}).get("entropy")
- if (
- saved_entropy is not None
- and abs(actual_entropy - saved_entropy) > 1.0
- ):
+ if saved_entropy is not None and abs(actual_entropy - saved_entropy) > 1.0:
issues.append(
f"Database structural anomaly (Entropy Δ: {abs(actual_entropy - saved_entropy):.2f})",
)
@@ -195,9 +190,7 @@ class IntegrityManager:
)
actual_size = full_path.stat().st_size
- is_critical = any(
- c in rel_path for c in ["identity", "config"]
- )
+ is_critical = any(c in rel_path for c in ["identity", "config"])
if is_critical:
issues.append(
diff --git a/meshchatx/src/backend/interface_port_check.py b/meshchatx/src/backend/interface_port_check.py
index 05233b19..6afb4e2c 100644
--- a/meshchatx/src/backend/interface_port_check.py
+++ b/meshchatx/src/backend/interface_port_check.py
@@ -15,7 +15,6 @@ import contextlib
import errno
import socket
-
_PORT_IN_USE_ERRNOS = {
errno.EADDRINUSE,
errno.EACCES,
@@ -28,7 +27,7 @@ def _normalize_host(host: str | None) -> str:
if host is None:
return ""
host = str(host).strip()
- if host == "" or host in {"*", "0.0.0.0", "::", "[::]"}:
+ if host == "" or host in {"*", "0.0.0.0", "::", "[::]"}: # noqa: S104
return ""
return host
@@ -64,7 +63,7 @@ def is_port_in_use(host: str | None, port, *, kind: str = "tcp") -> bool:
normalized = _normalize_host(host)
candidates: list[tuple[int, str]] = []
if normalized == "":
- candidates.append((socket.AF_INET, "0.0.0.0"))
+ candidates.append((socket.AF_INET, "0.0.0.0")) # noqa: S104
candidates.append((socket.AF_INET6, "::"))
else:
try:
@@ -110,7 +109,7 @@ def describe_port_conflict(
) -> str:
"""Build a user-facing message describing a port conflict."""
coerced_port = _coerce_port(port)
- host_label = _normalize_host(host) or "0.0.0.0"
+ host_label = _normalize_host(host) or "0.0.0.0" # noqa: S104
name_part = f' for interface "{interface_name}"' if interface_name else ""
proto = str(kind).upper()
if coerced_port is None:
diff --git a/meshchatx/src/backend/interfaces/WebsocketServerInterface.py b/meshchatx/src/backend/interfaces/WebsocketServerInterface.py
index 2b7a548e..f8a7184f 100644
--- a/meshchatx/src/backend/interfaces/WebsocketServerInterface.py
+++ b/meshchatx/src/backend/interfaces/WebsocketServerInterface.py
@@ -18,9 +18,7 @@ class WebsocketServerInterface(Interface):
RESTART_DELAY_SECONDS = 5
def __str__(self):
- return (
- f"WebsocketServerInterface[{self.name}/{self.listen_ip}:{self.listen_port}]"
- )
+ return f"WebsocketServerInterface[{self.name}/{self.listen_ip}:{self.listen_port}]"
def __init__(self, owner, configuration):
super().__init__()
diff --git a/meshchatx/src/backend/licenses_collector.py b/meshchatx/src/backend/licenses_collector.py
index eafa8d55..18d148fd 100644
--- a/meshchatx/src/backend/licenses_collector.py
+++ b/meshchatx/src/backend/licenses_collector.py
@@ -15,7 +15,7 @@ from datetime import UTC, datetime
from pathlib import Path
from typing import Any
-from packaging.requirements import Requirement
+from packaging.requirements import InvalidRequirement, Requirement
from packaging.utils import canonicalize_name
_ROOT_DIST_CANDIDATES = ("reticulum-meshchatx", "reticulum_meshchatx")
@@ -96,7 +96,7 @@ def _collect_python_transitive(root_names: tuple[str, ...]) -> list[dict[str, An
continue
try:
req = Requirement(req_str)
- except Exception:
+ except InvalidRequirement:
continue
if req.extras and not req.marker:
pass
@@ -125,7 +125,7 @@ def _python_roots_from_pyproject(repo_root: Path) -> tuple[str, ...]:
for line in deps:
try:
req = Requirement(line)
- except Exception:
+ except InvalidRequirement:
continue
names.append(req.name)
if not names:
@@ -297,7 +297,7 @@ def _try_pnpm_licenses(repo_root: Path) -> list[dict[str, Any]] | None:
if not isinstance(parsed, dict):
return None
rows = _flatten_pnpm_licenses_json(parsed)
- return rows if rows else None
+ return rows or None
def _flatten_pnpm_licenses_json(data: dict[str, Any]) -> list[dict[str, Any]]:
diff --git a/meshchatx/src/backend/lxmf_utils.py b/meshchatx/src/backend/lxmf_utils.py
index c69c9d03..eac38601 100644
--- a/meshchatx/src/backend/lxmf_utils.py
+++ b/meshchatx/src/backend/lxmf_utils.py
@@ -70,17 +70,13 @@ def is_user_facing_lxmf_payload(fields, content, title) -> bool:
return True
image = fields.get("image")
- if isinstance(image, dict) and (
- image.get("image_size") or image.get("image_bytes")
- ):
+ if isinstance(image, dict) and (image.get("image_size") or image.get("image_bytes")):
return True
if image is None and fields.get(LXMF_IMAGE_FIELD) is not None:
return True
audio = fields.get("audio")
- if isinstance(audio, dict) and (
- audio.get("audio_size") or audio.get("audio_bytes")
- ):
+ if isinstance(audio, dict) and (audio.get("audio_size") or audio.get("audio_bytes")):
return True
if audio is None and fields.get(LXMF_AUDIO_FIELD) is not None:
return True
@@ -114,10 +110,7 @@ def convert_lxmf_message_to_dict(
if field_type == LXMF.FIELD_FILE_ATTACHMENTS and isinstance(value, list):
file_attachments = []
for file_attachment in value:
- if (
- not isinstance(file_attachment, (list, tuple))
- or len(file_attachment) < 2
- ):
+ if not isinstance(file_attachment, (list, tuple)) or len(file_attachment) < 2:
continue
file_name = file_attachment[0]
file_data = file_attachment[1]
@@ -140,11 +133,7 @@ def convert_lxmf_message_to_dict(
fields["file_attachments"] = file_attachments
# handle image field
- if (
- field_type == LXMF.FIELD_IMAGE
- and isinstance(value, (list, tuple))
- and len(value) >= 2
- ):
+ if field_type == LXMF.FIELD_IMAGE and isinstance(value, (list, tuple)) and len(value) >= 2:
image_type = value[0]
image_data = value[1]
if isinstance(image_data, (bytes, bytearray)):
@@ -159,11 +148,7 @@ def convert_lxmf_message_to_dict(
}
# handle audio field
- if (
- field_type == LXMF.FIELD_AUDIO
- and isinstance(value, (list, tuple))
- and len(value) >= 2
- ):
+ if field_type == LXMF.FIELD_AUDIO and isinstance(value, (list, tuple)) and len(value) >= 2:
audio_mode = value[0]
audio_data = value[1]
if isinstance(audio_data, (bytes, bytearray)):
@@ -182,7 +167,7 @@ def convert_lxmf_message_to_dict(
fields["telemetry"] = Telemeter.from_packed(value)
# handle commands field
- if field_type == LXMF.FIELD_COMMANDS or field_type == 0x01:
+ if field_type in (LXMF.FIELD_COMMANDS, 1):
processed_commands = []
if isinstance(value, list):
for cmd in value:
@@ -211,9 +196,7 @@ def convert_lxmf_message_to_dict(
fields["reply_to"] = value.hex() if isinstance(value, bytes) else value
if field_type == 0x31:
fields["reply_quoted_content"] = (
- value.decode("utf-8", errors="replace")
- if isinstance(value, bytes)
- else value
+ value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
)
if field_type == LXMF_APP_EXTENSIONS_FIELD and isinstance(value, dict):
@@ -248,11 +231,7 @@ def convert_lxmf_message_to_dict(
val = message_fields[0x30]
reply_to_hash = val.hex() if isinstance(val, bytes) else val
- content = (
- lxmf_message.content.decode("utf-8", errors="replace")
- if lxmf_message.content
- else ""
- )
+ content = lxmf_message.content.decode("utf-8", errors="replace") if lxmf_message.content else ""
# auto-detect reply from content if not present
if not reply_to_hash and content and isinstance(content, str):
@@ -276,9 +255,7 @@ def convert_lxmf_message_to_dict(
"next_delivery_attempt",
None,
), # attribute may not exist yet
- "title": lxmf_message.title.decode("utf-8", errors="replace")
- if lxmf_message.title
- else "",
+ "title": lxmf_message.title.decode("utf-8", errors="replace") if lxmf_message.title else "",
"content": content,
"fields": fields,
"timestamp": lxmf_message.timestamp,
diff --git a/meshchatx/src/backend/map_manager.py b/meshchatx/src/backend/map_manager.py
index 65dfd813..671eab60 100644
--- a/meshchatx/src/backend/map_manager.py
+++ b/meshchatx/src/backend/map_manager.py
@@ -315,9 +315,7 @@ class MapManager:
)
return None
- tasks = [
- asyncio.create_task(download_tile(tile)) for tile in tiles_to_download
- ]
+ tasks = [asyncio.create_task(download_tile(tile)) for tile in tiles_to_download]
for coro in asyncio.as_completed(tasks):
if export_id in self._export_cancelled:
@@ -339,9 +337,7 @@ class MapManager:
(current_count / total_tiles) * 100,
)
- if len(batch_data) >= batch_size or (
- current_count == total_tiles and batch_data
- ):
+ if len(batch_data) >= batch_size or (current_count == total_tiles and batch_data):
try:
cursor.executemany(
"INSERT INTO tiles VALUES (?, ?, ?, ?)",
@@ -358,9 +354,7 @@ class MapManager:
n = 2.0**zoom
x = int((lon + 180.0) / 360.0 * n)
y = int(
- (1.0 - math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad))) / math.pi)
- / 2.0
- * n,
+ (1.0 - math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad))) / math.pi) / 2.0 * n,
)
return x, y
diff --git a/meshchatx/src/backend/meshchat_utils.py b/meshchatx/src/backend/meshchat_utils.py
index 9f1e232b..bfeda506 100644
--- a/meshchatx/src/backend/meshchat_utils.py
+++ b/meshchatx/src/backend/meshchat_utils.py
@@ -136,9 +136,9 @@ def parse_lxmf_display_name(
# Try manual parsing first to avoid LXMF library call.
if len(app_data_bytes) > 0:
- if (
- app_data_bytes[0] >= 0x90 and app_data_bytes[0] <= 0x9F
- ) or app_data_bytes[0] == 0xDC:
+ if (app_data_bytes[0] >= 0x90 and app_data_bytes[0] <= 0x9F) or app_data_bytes[
+ 0
+ ] == 0xDC:
with contextlib.suppress(Exception):
peer_data = msgpack.unpackb(app_data_bytes)
if isinstance(peer_data, list) and len(peer_data) >= 1:
diff --git a/meshchatx/src/backend/message_handler.py b/meshchatx/src/backend/message_handler.py
index 1d8edf30..4c46722d 100644
--- a/meshchatx/src/backend/message_handler.py
+++ b/meshchatx/src/backend/message_handler.py
@@ -123,7 +123,7 @@ class MessageHandler:
where_clauses = []
if folder_id is not None:
- if folder_id == 0 or folder_id == "0":
+ if folder_id in {0, "0"}:
# Special case: no folder (Uncategorized)
where_clauses.append("f.folder_id IS NULL")
else:
diff --git a/meshchatx/src/backend/nomadnet_downloader.py b/meshchatx/src/backend/nomadnet_downloader.py
index 3d750532..737e5898 100644
--- a/meshchatx/src/backend/nomadnet_downloader.py
+++ b/meshchatx/src/backend/nomadnet_downloader.py
@@ -36,11 +36,7 @@ def get_cached_active_link(destination_hash: bytes):
def sweep_stale_links():
"""Evict all non-ACTIVE links from the global cache."""
with _nomadnet_links_lock:
- stale = [
- k
- for k, v in nomadnet_cached_links.items()
- if v.status is not RNS.Link.ACTIVE
- ]
+ stale = [k for k, v in nomadnet_cached_links.items() if v.status is not RNS.Link.ACTIVE]
for k in stale:
del nomadnet_cached_links[k]
@@ -185,9 +181,7 @@ class NomadnetDownloader:
timeout_after_seconds = time.time() + link_establishment_timeout
- while (
- link.status is not RNS.Link.ACTIVE and time.time() < timeout_after_seconds
- ):
+ while link.status is not RNS.Link.ACTIVE and time.time() < timeout_after_seconds:
if self.is_cancelled:
return
await asyncio.sleep(_POLL_INTERVAL_S)
@@ -307,11 +301,7 @@ class NomadnetFileDownloader(NomadnetDownloader):
self.on_file_download_success(file_name, file_data)
return
- if (
- isinstance(response, list)
- and len(response) > 1
- and isinstance(response[1], dict)
- ):
+ if isinstance(response, list) and len(response) > 1 and isinstance(response[1], dict):
file_data: bytes = response[0]
metadata: dict = response[1]
diff --git a/meshchatx/src/backend/page_node.py b/meshchatx/src/backend/page_node.py
index 7545e4e6..55d436e8 100644
--- a/meshchatx/src/backend/page_node.py
+++ b/meshchatx/src/backend/page_node.py
@@ -336,8 +336,7 @@ class PageNode:
return sorted(
f
for f in os.listdir(self.pages_dir)
- if os.path.isfile(os.path.join(self.pages_dir, f))
- and is_allowed_page_filename(f)
+ if os.path.isfile(os.path.join(self.pages_dir, f)) and is_allowed_page_filename(f)
)
def get_page_content(self, name):
diff --git a/meshchatx/src/backend/page_node_manager.py b/meshchatx/src/backend/page_node_manager.py
index a16024e1..c56bb4c8 100644
--- a/meshchatx/src/backend/page_node_manager.py
+++ b/meshchatx/src/backend/page_node_manager.py
@@ -79,8 +79,7 @@ class PageNodeManager:
if node.running:
return node.get_destination_hash()
- dest_hash = node.setup()
- return dest_hash
+ return node.setup()
def stop_node(self, node_id):
"""Stop serving for a specific node."""
diff --git a/meshchatx/src/backend/persistent_log_handler.py b/meshchatx/src/backend/persistent_log_handler.py
index 5c86a603..b95059e7 100644
--- a/meshchatx/src/backend/persistent_log_handler.py
+++ b/meshchatx/src/backend/persistent_log_handler.py
@@ -64,9 +64,7 @@ class PersistentLogHandler(logging.Handler):
self._error_events.append(now_mono)
# Periodically flush to database if available
- if self.database and (
- time.time() - self.last_flush_time > self.flush_interval
- ):
+ if self.database and (time.time() - self.last_flush_time > self.flush_interval):
self._flush_to_db()
except Exception:
@@ -228,9 +226,7 @@ class PersistentLogHandler(logging.Handler):
if level:
logs = [log for log in logs if log["level"] == level]
if is_anomaly is not None:
- logs = [
- log for log in logs if log["is_anomaly"] == (1 if is_anomaly else 0)
- ]
+ logs = [log for log in logs if log["is_anomaly"] == (1 if is_anomaly else 0)]
# Sort descending
logs.sort(key=lambda x: x["timestamp"], reverse=True)
diff --git a/meshchatx/src/backend/recovery/crash_recovery.py b/meshchatx/src/backend/recovery/crash_recovery.py
index 953c2a76..625bbf03 100644
--- a/meshchatx/src/backend/recovery/crash_recovery.py
+++ b/meshchatx/src/backend/recovery/crash_recovery.py
@@ -156,7 +156,7 @@ class CrashRecovery:
cause_counts = {r["diagnosed_cause"]: r["count"] for r in freq_rows}
weights = {}
- for key, default_prior in _DEFAULT_PRIORS.items():
+ for key in _DEFAULT_PRIORS:
desc = self._cause_key_to_description(key)
count = cause_counts.get(desc, 0)
alpha = 1.0 + count
@@ -383,9 +383,7 @@ class CrashRecovery:
"no_table_config": "no such table: config" in error_msg,
"in_memory_db": diagnosis.get("db_type") == "memory",
"corrupt_in_msg": "corrupt" in error_msg or "malformed" in error_msg,
- "async_in_msg": any(
- x in error_msg for x in ["asyncio", "event loop", "runtimeerror"]
- ),
+ "async_in_msg": any(x in error_msg for x in ["asyncio", "event loop", "runtimeerror"]),
"no_loop_in_msg": "no current event loop" in error_msg
or "no running event loop" in error_msg,
"low_mem": diagnosis.get("low_memory", False),
@@ -394,11 +392,11 @@ class CrashRecovery:
"lxmf_in_msg": "lxmf" in error_msg or "lxmr" in error_msg,
"identity_in_msg": "identity" in error_msg or "private key" in error_msg,
"no_interfaces": diagnosis.get("active_interfaces", 0) == 0,
- "old_python": py_version.major < 3
- or (py_version.major == 3 and py_version.minor < 10),
+ "old_python": py_version.major < 3 or (py_version.major == 3 and py_version.minor < 10),
"legacy_kernel": "linux" in platform.system().lower()
- and (lambda m: m is not None and float(m.group(1)) < 4.0)(
- re.search(r"(\d+\.\d+)", platform.release()),
+ and (
+ (_m := re.search(r"(\d+\.\d+)", platform.release())) is not None
+ and float(_m.group(1)) < 4.0
),
"attribute_error": "attributeerror" in error_type,
}
@@ -523,7 +521,7 @@ class CrashRecovery:
entropy = sum(h(q) for q in q_vec)
# Systemic Divergence: How 'surprising' this state is compared to ideal
- divergence = sum(kl_div(q, p) for q, p in zip(q_vec, p_vec))
+ divergence = sum(kl_div(q, p) for q, p in zip(q_vec, p_vec, strict=False))
return entropy, divergence
diff --git a/meshchatx/src/backend/rncp_handler.py b/meshchatx/src/backend/rncp_handler.py
index fb095a01..81202520 100644
--- a/meshchatx/src/backend/rncp_handler.py
+++ b/meshchatx/src/backend/rncp_handler.py
@@ -178,9 +178,7 @@ class RNCPHandler:
if transfer_id in self.active_transfers:
self.active_transfers[transfer_id]["status"] = "completed"
- self.active_transfers[transfer_id]["saved_path"] = (
- saved_filename
- )
+ self.active_transfers[transfer_id]["saved_path"] = saved_filename
self.active_transfers[transfer_id]["filename"] = filename
self._emit_receive_event(
{
@@ -279,9 +277,7 @@ class RNCPHandler:
RNS.Transport.request_path(destination_hash)
timeout_after = time.time() + timeout
- while (
- not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after
- ):
+ while not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after:
await asyncio.sleep(0.1)
if not RNS.Transport.has_path(destination_hash):
@@ -373,9 +369,7 @@ class RNCPHandler:
RNS.Transport.request_path(destination_hash)
timeout_after = time.time() + timeout
- while (
- not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after
- ):
+ while not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after:
await asyncio.sleep(0.1)
if not RNS.Transport.has_path(destination_hash):
diff --git a/meshchatx/src/backend/rnpath_handler.py b/meshchatx/src/backend/rnpath_handler.py
index fbd213ba..94fda110 100644
--- a/meshchatx/src/backend/rnpath_handler.py
+++ b/meshchatx/src/backend/rnpath_handler.py
@@ -9,10 +9,10 @@ class RNPathHandler:
def get_path_table(
self,
- max_hops: int = None,
- search: str = None,
- interface: str = None,
- hops: int = None,
+ max_hops: int | None = None,
+ search: str | None = None,
+ interface: str | None = None,
+ hops: int | None = None,
page: int = 1,
limit: int = 0,
):
@@ -71,18 +71,10 @@ class RNPathHandler:
total = len(formatted_table)
responsive_count = len(
- [
- e
- for e in formatted_table
- if e["state"] == RNS.Transport.STATE_RESPONSIVE
- ],
+ [e for e in formatted_table if e["state"] == RNS.Transport.STATE_RESPONSIVE],
)
unresponsive_count = len(
- [
- e
- for e in formatted_table
- if e["state"] == RNS.Transport.STATE_UNRESPONSIVE
- ],
+ [e for e in formatted_table if e["state"] == RNS.Transport.STATE_UNRESPONSIVE],
)
# Pagination
diff --git a/meshchatx/src/backend/rnpath_trace_handler.py b/meshchatx/src/backend/rnpath_trace_handler.py
index 44d6a018..9b3215c6 100644
--- a/meshchatx/src/backend/rnpath_trace_handler.py
+++ b/meshchatx/src/backend/rnpath_trace_handler.py
@@ -51,11 +51,7 @@ class RNPathTraceHandler:
local_hash = "unknown"
if self.identity and hasattr(self.identity, "hash"):
local_hash = self.identity.hash.hex()
- elif (
- self.reticulum
- and hasattr(self.reticulum, "identity")
- and self.reticulum.identity
- ):
+ elif self.reticulum and hasattr(self.reticulum, "identity") and self.reticulum.identity:
local_hash = self.reticulum.identity.hash.hex()
path.append({"type": "local", "hash": local_hash, "name": "Local Node"})
diff --git a/meshchatx/src/backend/rnprobe_handler.py b/meshchatx/src/backend/rnprobe_handler.py
index 5dc54422..89440445 100644
--- a/meshchatx/src/backend/rnprobe_handler.py
+++ b/meshchatx/src/backend/rnprobe_handler.py
@@ -28,19 +28,15 @@ class RNProbeHandler:
app_name, aspects = RNS.Destination.app_and_aspects_from_name(full_name)
except Exception as e:
msg = f"Invalid destination name: {e}"
- raise ValueError(msg)
+ raise ValueError(msg) from e
if not RNS.Transport.has_path(destination_hash):
RNS.Transport.request_path(destination_hash)
timeout_after = time.time() + (
- timeout
- or self.DEFAULT_TIMEOUT
- + self.reticulum.get_first_hop_timeout(destination_hash)
+ timeout or self.DEFAULT_TIMEOUT + self.reticulum.get_first_hop_timeout(destination_hash)
)
- while (
- not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after
- ):
+ while not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after:
await asyncio.sleep(0.1)
if not RNS.Transport.has_path(destination_hash):
@@ -65,10 +61,13 @@ class RNProbeHandler:
try:
probe = RNS.Packet(request_destination, os.urandom(size))
+ except OSError as e:
+ raise ValueError(f"Failed to build probe packet: {e!s}") from e
+ try:
probe.pack()
- except OSError:
+ except OSError as e:
msg = f"Probe packet size of {len(probe.raw)} bytes exceeds MTU of {RNS.Reticulum.MTU} bytes"
- raise ValueError(msg)
+ raise ValueError(msg) from e
receipt = probe.send()
sent += 1
@@ -80,12 +79,9 @@ class RNProbeHandler:
timeout_after = time.time() + (
timeout
- or self.DEFAULT_TIMEOUT
- + self.reticulum.get_first_hop_timeout(destination_hash)
+ or self.DEFAULT_TIMEOUT + self.reticulum.get_first_hop_timeout(destination_hash)
)
- while (
- receipt.status == RNS.PacketReceipt.SENT and time.time() < timeout_after
- ):
+ while receipt.status == RNS.PacketReceipt.SENT and time.time() < timeout_after:
await asyncio.sleep(0.1)
result: dict = {
diff --git a/meshchatx/src/backend/rnstatus_handler.py b/meshchatx/src/backend/rnstatus_handler.py
index b65cb998..fea92585 100644
--- a/meshchatx/src/backend/rnstatus_handler.py
+++ b/meshchatx/src/backend/rnstatus_handler.py
@@ -51,7 +51,7 @@ def fmt_packet_count(value: Any) -> str | None:
x = float(value)
except (TypeError, ValueError):
return str(value)
- return f"{int(round(x)):,}"
+ return f"{round(x):,}"
def fmt_percentage(value: Any) -> str | None:
@@ -195,10 +195,8 @@ class RNStatusHandler:
for ifstat in interfaces:
name = ifstat.get("name", "")
- if (
- name.startswith("LocalInterface[")
- or name.startswith("TCPInterface[Client")
- or name.startswith("BackboneInterface[Client on")
+ if name.startswith(
+ ("LocalInterface[", "TCPInterface[Client", "BackboneInterface[Client on")
):
continue
diff --git a/meshchatx/src/backend/sticker_pack_utils.py b/meshchatx/src/backend/sticker_pack_utils.py
index 31ddce74..3e984ea0 100644
--- a/meshchatx/src/backend/sticker_pack_utils.py
+++ b/meshchatx/src/backend/sticker_pack_utils.py
@@ -124,9 +124,7 @@ def validate_pack_document(data: object) -> dict:
out_stickers.append(
{
"name": item.get("name") if isinstance(item.get("name"), str) else None,
- "emoji": item.get("emoji")
- if isinstance(item.get("emoji"), str)
- else None,
+ "emoji": item.get("emoji") if isinstance(item.get("emoji"), str) else None,
"image_type": item.get("image_type"),
"image_bytes_b64": b64.strip(),
},
@@ -137,9 +135,7 @@ def validate_pack_document(data: object) -> dict:
"short_name": sanitize_pack_short_name(pack_meta.get("short_name")),
"description": sanitize_pack_description(pack_meta.get("description")),
"pack_type": sanitize_pack_type(pack_meta.get("type")),
- "author": pack_meta.get("author")
- if isinstance(pack_meta.get("author"), str)
- else None,
+ "author": pack_meta.get("author") if isinstance(pack_meta.get("author"), str) else None,
"is_strict": bool(pack_meta.get("is_strict", True)),
},
"stickers": out_stickers,
diff --git a/meshchatx/src/backend/sticker_utils.py b/meshchatx/src/backend/sticker_utils.py
index 6116592d..baae94c2 100644
--- a/meshchatx/src/backend/sticker_utils.py
+++ b/meshchatx/src/backend/sticker_utils.py
@@ -260,7 +260,7 @@ def parse_tgs(data: bytes) -> dict:
if width <= 0 or height <= 0 or fps <= 0 or out_point <= in_point:
msg = "invalid_tgs_metadata"
raise ValueError(msg)
- duration_ms = int(round(((out_point - in_point) / fps) * 1000.0))
+ duration_ms = round(((out_point - in_point) / fps) * 1000.0)
return {
"width": width,
"height": height,
@@ -270,9 +270,7 @@ def parse_tgs(data: bytes) -> dict:
}
-def _ebml_read_vint(
- buf: bytes, pos: int, *, mask_marker: bool = True
-) -> tuple[int, int] | None:
+def _ebml_read_vint(buf: bytes, pos: int, *, mask_marker: bool = True) -> tuple[int, int] | None:
"""Read an EBML variable-length integer at ``pos``; returns ``(value, next_pos)``."""
if pos >= len(buf):
return None
@@ -364,15 +362,11 @@ def parse_webm(data: bytes) -> dict:
if f_id == 0x83:
track_type = _ebml_read_uint(raw, fb, fe)
elif f_id == 0x86:
- track_codec = raw[fb:fe].decode(
- "ascii", errors="replace"
- )
+ track_codec = raw[fb:fe].decode("ascii", errors="replace")
elif f_id == 0x23E383:
track_def_dur = _ebml_read_uint(raw, fb, fe)
elif f_id == 0xE0:
- for v_id, vb, ve in _ebml_iter_elements(
- raw, fb, fe
- ):
+ for v_id, vb, ve in _ebml_iter_elements(raw, fb, fe):
if v_id == 0xB0:
t_w = _ebml_read_uint(raw, vb, ve)
elif v_id == 0xBA:
@@ -388,9 +382,7 @@ def parse_webm(data: bytes) -> dict:
elif seg_id == 0x1549A966:
for inf_id, ib, ie in _ebml_iter_elements(raw, sb, se):
if inf_id == 0x2AD7B1:
- timecode_scale = (
- _ebml_read_uint(raw, ib, ie) or timecode_scale
- )
+ timecode_scale = _ebml_read_uint(raw, ib, ie) or timecode_scale
elif inf_id == 0x4489:
d = _ebml_read_float(raw, ib, ie)
if d is not None:
@@ -399,7 +391,7 @@ def parse_webm(data: bytes) -> dict:
msg = "invalid_webm_no_video"
raise ValueError(msg)
if duration_ticks > 0:
- duration_ms = int(round(duration_ticks * timecode_scale / 1_000_000.0))
+ duration_ms = round(duration_ticks * timecode_scale / 1_000_000.0)
fps = 0.0
if track_default_duration_ns:
fps = 1_000_000_000.0 / float(track_default_duration_ns[0])
@@ -420,7 +412,7 @@ def _validate_dimensions_telegram_static(width: int, height: int) -> None:
if width > STICKER_CANVAS or height > STICKER_CANVAS:
msg = "dimensions_too_large"
raise ValueError(msg)
- if width != STICKER_CANVAS and height != STICKER_CANVAS:
+ if STICKER_CANVAS not in (width, height):
msg = "dimensions_not_512_on_one_side"
raise ValueError(msg)
diff --git a/meshchatx/src/backend/telemetry_utils.py b/meshchatx/src/backend/telemetry_utils.py
index 324f23d9..a05109ee 100644
--- a/meshchatx/src/backend/telemetry_utils.py
+++ b/meshchatx/src/backend/telemetry_utils.py
@@ -66,12 +66,12 @@ class Telemeter:
):
try:
return [
- struct.pack("!i", int(round(latitude * 1e6))),
- struct.pack("!i", int(round(longitude * 1e6))),
- struct.pack("!i", int(round(altitude * 1e2))),
- struct.pack("!I", int(round(speed * 1e2))),
- struct.pack("!i", int(round(bearing * 1e2))),
- struct.pack("!H", int(round(accuracy * 1e2))),
+ struct.pack("!i", round(latitude * 1e6)),
+ struct.pack("!i", round(longitude * 1e6)),
+ struct.pack("!i", round(altitude * 1e2)),
+ struct.pack("!I", round(speed * 1e2)),
+ struct.pack("!i", round(bearing * 1e2)),
+ struct.pack("!H", round(accuracy * 1e2)),
int(last_update) if last_update is not None else int(time.time()),
]
except Exception:
diff --git a/meshchatx/src/backend/telephone_manager.py b/meshchatx/src/backend/telephone_manager.py
index 9a21b3f9..b8f33f35 100644
--- a/meshchatx/src/backend/telephone_manager.py
+++ b/meshchatx/src/backend/telephone_manager.py
@@ -60,9 +60,7 @@ class TelephoneManager:
self.storage_dir = storage_dir
self.db = db
self.get_name_for_identity_hash = None
- self.recordings_dir = (
- os.path.join(storage_dir, "recordings") if storage_dir else None
- )
+ self.recordings_dir = os.path.join(storage_dir, "recordings") if storage_dir else None
if self.recordings_dir:
os.makedirs(self.recordings_dir, exist_ok=True)
@@ -445,8 +443,7 @@ class TelephoneManager:
# to ensure the UI has something to show (either active_call or initiation_status)
for _ in range(40): # Max 4 seconds of defensive waiting
if self.telephone and (
- self.telephone.active_call
- or self.telephone.call_status in [0, 1, 3, 6]
+ self.telephone.active_call or self.telephone.call_status in [0, 1, 3, 6]
):
break
await asyncio.sleep(self._status_poll_interval_s)
@@ -472,7 +469,7 @@ class TelephoneManager:
# Still call the internal method just in case it does something useful
try:
self.telephone.mute_transmit()
- except Exception: # noqa: S110
+ except Exception:
pass
self.transmit_muted = True
@@ -492,7 +489,7 @@ class TelephoneManager:
# Still call the internal method just in case
try:
self.telephone.unmute_transmit()
- except Exception: # noqa: S110
+ except Exception:
pass
self.transmit_muted = False
@@ -509,7 +506,7 @@ class TelephoneManager:
# Still call the internal method just in case
try:
self.telephone.mute_receive()
- except Exception: # noqa: S110
+ except Exception:
pass
self.receive_muted = True
@@ -529,7 +526,7 @@ class TelephoneManager:
# Still call the internal method just in case
try:
self.telephone.unmute_receive()
- except Exception: # noqa: S110
+ except Exception:
pass
self.receive_muted = False
diff --git a/meshchatx/src/backend/translator_handler.py b/meshchatx/src/backend/translator_handler.py
index 2be09dc8..787beba3 100644
--- a/meshchatx/src/backend/translator_handler.py
+++ b/meshchatx/src/backend/translator_handler.py
@@ -261,7 +261,9 @@ class TranslatorHandler:
if detected_lang:
source_lang = detected_lang
else:
- msg = "Could not auto-detect language. Please select a source language manually."
+ msg = (
+ "Could not auto-detect language. Please select a source language manually."
+ )
raise ValueError(msg)
else:
msg = (
@@ -308,7 +310,7 @@ class TranslatorHandler:
}
except Exception as e:
msg = f"Argos Translate error: {e}"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
def _translate_argos_cli(
self,
@@ -349,7 +351,7 @@ class TranslatorHandler:
target_lang,
text,
]
- result = subprocess.run(args, capture_output=True, text=True, check=True) # noqa: S603
+ result = subprocess.run(args, capture_output=True, text=True, check=True)
translated_text = result.stdout.strip()
if not translated_text:
msg = "Translation returned empty result"
@@ -361,16 +363,12 @@ class TranslatorHandler:
"source": "argos",
}
except subprocess.CalledProcessError as e:
- error_msg = (
- e.stderr.decode()
- if isinstance(e.stderr, bytes)
- else (e.stderr or str(e))
- )
+ error_msg = e.stderr.decode() if isinstance(e.stderr, bytes) else (e.stderr or str(e))
msg = f"Argos Translate CLI error: {error_msg}"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
except Exception as e:
msg = f"Argos Translate CLI error: {e!s}"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
def _detect_language(self, text: str) -> str | None:
if not self.has_argos_lib:
@@ -398,7 +396,7 @@ class TranslatorHandler:
return languages
try:
- result = subprocess.run( # noqa: S603
+ result = subprocess.run(
[argospm, "list"],
capture_output=True,
text=True,
@@ -444,7 +442,7 @@ class TranslatorHandler:
raise RuntimeError(msg)
try:
- result = subprocess.run( # noqa: S603
+ result = subprocess.run(
[argospm, "install", package_name],
capture_output=True,
text=True,
@@ -456,12 +454,12 @@ class TranslatorHandler:
"message": f"Successfully installed {package_name}",
"output": result.stdout,
}
- except subprocess.TimeoutExpired:
+ except subprocess.TimeoutExpired as e:
msg = f"Installation of {package_name} timed out after 5 minutes"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
except subprocess.CalledProcessError as e:
msg = f"Failed to install {package_name}: {e.stderr or str(e)}"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
except Exception as e:
msg = f"Error installing {package_name}: {e!s}"
- raise RuntimeError(msg)
+ raise RuntimeError(msg) from e
diff --git a/meshchatx/src/backend/voicemail_manager.py b/meshchatx/src/backend/voicemail_manager.py
index 509f809f..c8fb490a 100644
--- a/meshchatx/src/backend/voicemail_manager.py
+++ b/meshchatx/src/backend/voicemail_manager.py
@@ -140,7 +140,7 @@ class VoicemailManager:
f"Voicemail: Generating greeting with command: {' '.join(cmd)}",
RNS.LOG_DEBUG,
)
- subprocess.run(cmd, check=True) # noqa: S603
+ subprocess.run(cmd, check=True)
return self.convert_to_greeting(wav_path)
finally:
diff --git a/meshchatx/src/backend/web_audio_bridge.py b/meshchatx/src/backend/web_audio_bridge.py
index 828ae0ef..c96b4fb1 100644
--- a/meshchatx/src/backend/web_audio_bridge.py
+++ b/meshchatx/src/backend/web_audio_bridge.py
@@ -48,16 +48,14 @@ class WebAudioSource(LocalSource):
def push_pcm(self, pcm_bytes: bytes):
try:
- samples = (
- np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
- )
+ samples = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
if samples.size == 0:
return
samples = samples.reshape(-1, 1)
frame = self.codec.encode(samples)
if self.sink and self.sink.can_receive(from_source=self):
self.sink.handle_frame(frame, self)
- except Exception as exc: # noqa: BLE001
+ except Exception as exc:
RNS.log(f"WebAudioSource: failed to push pcm: {exc}", RNS.LOG_ERROR)
@@ -80,7 +78,7 @@ class WebAudioSink(LocalSink):
else:
pcm = frame
self.loop.call_soon_threadsafe(asyncio.create_task, self.send_bytes(pcm))
- except Exception as exc: # noqa: BLE001
+ except Exception as exc:
RNS.log(f"WebAudioSink: failed to handle frame: {exc}", RNS.LOG_ERROR)
@@ -188,7 +186,7 @@ class WebAudioBridge:
tele.audio_input = self.tx_source
if tele.transmit_mixer and not tele.transmit_mixer.should_run:
tele.transmit_mixer.start()
- except Exception as exc: # noqa: BLE001
+ except Exception as exc:
RNS.log(
f"WebAudioBridge: failed to swap transmit path: {exc}",
RNS.LOG_ERROR,
@@ -214,7 +212,7 @@ class WebAudioBridge:
sink=self.rx_tee,
)
tele.receive_pipeline.start()
- except Exception as exc: # noqa: BLE001
+ except Exception as exc:
RNS.log(f"WebAudioBridge: failed to tee receive path: {exc}", RNS.LOG_ERROR)
def _restore_host_audio(self):
diff --git a/scripts/argos_translate.py b/scripts/argos_translate.py
index 40e96022..eb26c3d5 100755
--- a/scripts/argos_translate.py
+++ b/scripts/argos_translate.py
@@ -93,17 +93,13 @@ def ensure_package_installed(from_code, to_code):
if pkg_to_install:
print_info(f"Downloading package: {pkg_to_install}")
argostranslate.package.install_from_path(pkg_to_install.download())
- print_success(
- f"Successfully installed package: {from_code} -> {to_code}"
- )
+ print_success(f"Successfully installed package: {from_code} -> {to_code}")
# Refresh installed languages
installed = argostranslate.translate.get_installed_languages()
installed_dict = {lang.code: lang for lang in installed}
else:
- print_error(
- f"Could not find a translation package for {from_code} -> {to_code}"
- )
+ print_error(f"Could not find a translation package for {from_code} -> {to_code}")
sys.exit(1)
except Exception as e:
print_error(f"Failed to install language package: {e}")
@@ -160,16 +156,16 @@ def translate_dict(data, translate_func, target_name=None):
if k == "_languageName" and target_name:
new_dict[k] = target_name
continue
- elif k == "_languageName":
+ if k == "_languageName":
# Keep original if no target name provided
new_dict[k] = v
continue
new_dict[k] = translate_dict(v, translate_func, target_name)
return new_dict
- elif isinstance(data, list):
+ if isinstance(data, list):
return [translate_dict(item, translate_func, target_name) for item in data]
- elif isinstance(data, str):
+ if isinstance(data, str):
if not data.strip():
return data
@@ -178,9 +174,7 @@ def translate_dict(data, translate_func, target_name=None):
translated_temp = translate_func(temp_text)
return restore_vars_from_tokens(translated_temp, vars_found)
except Exception as e:
- print_warning(
- f"Failed to translate '{data}': {e}. Falling back to original."
- )
+ print_warning(f"Failed to translate '{data}': {e}. Falling back to original.")
return data
else:
return data
@@ -190,9 +184,7 @@ def main():
parser = argparse.ArgumentParser(
description="Translate JSON localization files using Argos Translate."
)
- parser.add_argument(
- "--from", dest="from_lang", help="Source language code (e.g. 'en')"
- )
+ parser.add_argument("--from", dest="from_lang", help="Source language code (e.g. 'en')")
parser.add_argument("--to", dest="to_lang", help="Target language code (e.g. 'zh')")
parser.add_argument("--input", dest="input_file", help="Path to input JSON file")
parser.add_argument("--output", dest="output_file", help="Path to output JSON file")
@@ -229,7 +221,7 @@ def main():
# Load JSON
try:
- with open(input_file, "r", encoding="utf-8") as f:
+ with open(input_file, encoding="utf-8") as f:
source_data = json.load(f)
except json.JSONDecodeError as e:
print_error(f"Invalid JSON in input file: {e}")
@@ -240,9 +232,7 @@ def main():
# Get Translator
translate_func = get_translation_func(from_lang, to_lang)
- print_info(
- "Starting translation. This may take a moment depending on the file size..."
- )
+ print_info("Starting translation. This may take a moment depending on the file size...")
translated_data = translate_dict(source_data, translate_func, target_name)
# Ensure output directory exists
@@ -253,7 +243,7 @@ def main():
with open(output_file, "w", encoding="utf-8") as f:
json.dump(translated_data, f, ensure_ascii=False, indent=4)
f.write("\n")
- except IOError as e:
+ except OSError as e:
print_error(f"Could not write to output file: {e}")
sys.exit(1)
diff --git a/scripts/build/fetch_reticulum_manual.py b/scripts/build/fetch_reticulum_manual.py
index 6ea2c778..6cb6e1cf 100755
--- a/scripts/build/fetch_reticulum_manual.py
+++ b/scripts/build/fetch_reticulum_manual.py
@@ -61,11 +61,11 @@ def _is_truthy(value: str | None) -> bool:
def _download(url: str, timeout: float) -> bytes:
logging.info("Downloading Reticulum manual from %s", url)
- req = urllib.request.Request( # noqa: S310 - URL is constrained to known sources
+ req = urllib.request.Request(
url,
headers={"User-Agent": "meshchatx-build-script"},
)
- with urllib.request.urlopen(req, timeout=timeout) as response: # noqa: S310
+ with urllib.request.urlopen(req, timeout=timeout) as response:
return response.read()
@@ -240,9 +240,7 @@ def main(argv: list[str] | None = None) -> int:
)
if _is_truthy(os.environ.get("MESHCHATX_SKIP_DOCS_FETCH")):
- logging.info(
- "MESHCHATX_SKIP_DOCS_FETCH is set; skipping Reticulum manual fetch."
- )
+ logging.info("MESHCHATX_SKIP_DOCS_FETCH is set; skipping Reticulum manual fetch.")
return 0
sources: list[str] = []
diff --git a/scripts/build_community_interfaces_json.py b/scripts/build_community_interfaces_json.py
index 240b97d3..502cd24e 100644
--- a/scripts/build_community_interfaces_json.py
+++ b/scripts/build_community_interfaces_json.py
@@ -2,7 +2,6 @@
# SPDX-License-Identifier: 0BSD
"""Emit meshchatx/src/backend/data/community_interfaces.json from the directory API or a local export."""
-# ruff: noqa: T201
import argparse
import json
diff --git a/scripts/ci/slsa-predicate.py b/scripts/ci/slsa-predicate.py
index 04affe42..a269d32b 100644
--- a/scripts/ci/slsa-predicate.py
+++ b/scripts/ci/slsa-predicate.py
@@ -14,12 +14,10 @@ def _source_uri() -> str:
server = (
os.environ.get("GITHUB_SERVER_URL") or os.environ.get("GITEA_SERVER_URL") or ""
).rstrip("/")
- repo = (
- os.environ.get("GITHUB_REPOSITORY") or os.environ.get("GITEA_REPOSITORY") or ""
- )
+ repo = os.environ.get("GITHUB_REPOSITORY") or os.environ.get("GITEA_REPOSITORY") or ""
if not server or not repo:
return ""
- if server.startswith("https://") or server.startswith("http://"):
+ if server.startswith(("https://", "http://")):
return f"git+{server}/{repo}.git"
return f"git+https://{server}/{repo}.git"
@@ -31,9 +29,7 @@ def _build_type() -> str:
server = (
os.environ.get("GITHUB_SERVER_URL") or os.environ.get("GITEA_SERVER_URL") or ""
).rstrip("/")
- repo = (
- os.environ.get("GITHUB_REPOSITORY") or os.environ.get("GITEA_REPOSITORY") or ""
- )
+ repo = os.environ.get("GITHUB_REPOSITORY") or os.environ.get("GITEA_REPOSITORY") or ""
if server and repo:
return f"{server}/{repo}/.gitea/workflows/build.yml"
return "https://slsa.dev/provenance/v1"
diff --git a/tests/backend/benchmarking_utils.py b/tests/backend/benchmarking_utils.py
index e9114ec1..8eb46b49 100644
--- a/tests/backend/benchmarking_utils.py
+++ b/tests/backend/benchmarking_utils.py
@@ -21,7 +21,9 @@ class BenchmarkResult:
self.memory_delta_mb = memory_delta_mb
def __repr__(self):
- return f"<BenchmarkResult {self.name}: {self.duration_ms:.2f}ms, {self.memory_delta_mb:.2f}MB>"
+ return (
+ f"<BenchmarkResult {self.name}: {self.duration_ms:.2f}ms, {self.memory_delta_mb:.2f}MB>"
+ )
def benchmark(name=None, iterations=1):
diff --git a/tests/backend/test_audio_codec.py b/tests/backend/test_audio_codec.py
index f7fa51a1..5a02d65d 100644
--- a/tests/backend/test_audio_codec.py
+++ b/tests/backend/test_audio_codec.py
@@ -38,9 +38,7 @@ def _build_wav_pcm16(
wf.setframerate(samplerate)
frames = bytearray()
for i in range(n_samples):
- sample = int(
- 0.3 * 32767 * math.sin(2 * math.pi * frequency * (i / samplerate))
- )
+ sample = int(0.3 * 32767 * math.sin(2 * math.pi * frequency * (i / samplerate)))
for _ in range(channels):
frames.extend(struct.pack("<h", sample))
wf.writeframes(bytes(frames))
@@ -194,9 +192,7 @@ def test_encode_pcm_to_ogg_opus_preserves_duration(duration_seconds):
sr = 48000
n = int(sr * duration_seconds)
t = np.arange(n, dtype=np.float32) / sr
- samples = (
- (0.3 * np.sin(2 * math.pi * 440.0 * t)).astype(np.float32).reshape(-1, 1)
- )
+ samples = (0.3 * np.sin(2 * math.pi * 440.0 * t)).astype(np.float32).reshape(-1, 1)
audio_codec.encode_pcm_to_ogg_opus(samples, sr, 1, out)
encoded = _ogg_opus_duration_seconds(out)
assert abs(encoded - duration_seconds) < 0.001, (
@@ -219,9 +215,7 @@ def test_encode_pcm_to_ogg_opus_audio_profile_keeps_stereo():
t = np.arange(n, dtype=np.float32) / sr
samples[:, 0] = 0.3 * np.sin(2 * math.pi * 440.0 * t)
samples[:, 1] = 0.3 * np.sin(2 * math.pi * 660.0 * t)
- audio_codec.encode_pcm_to_ogg_opus(
- samples, sr, 2, out, profile=Opus.PROFILE_AUDIO_MAX
- )
+ audio_codec.encode_pcm_to_ogg_opus(samples, sr, 2, out, profile=Opus.PROFILE_AUDIO_MAX)
with open(out, "rb") as f:
data = f.read()
head = data.find(b"OpusHead")
diff --git a/tests/backend/test_auto_propagation.py b/tests/backend/test_auto_propagation.py
index e9538a10..22e9a341 100644
--- a/tests/backend/test_auto_propagation.py
+++ b/tests/backend/test_auto_propagation.py
@@ -56,9 +56,7 @@ async def test_auto_propagation_logic():
patch.object(manager, "_wait_for_path", return_value=True),
patch.object(manager, "_probe_propagation_sync", return_value=True),
):
- mock_hops.side_effect = lambda dh: (
- 1 if dh == bytes.fromhex(_VALID_HASH_A) else 3
- )
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex(_VALID_HASH_A) else 3
await manager.check_and_update_propagation_node()
@@ -70,9 +68,7 @@ async def test_auto_propagation_logic():
_VALID_HASH_A,
)
- config.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
- _VALID_HASH_B
- )
+ config.lxmf_preferred_propagation_node_destination_hash.get.return_value = _VALID_HASH_B
app.set_active_propagation_node.reset_mock()
with (
@@ -81,9 +77,7 @@ async def test_auto_propagation_logic():
patch.object(manager, "_wait_for_path", return_value=True),
patch.object(manager, "_probe_propagation_sync", side_effect=[False, True]),
):
- mock_hops.side_effect = lambda dh: (
- 1 if dh == bytes.fromhex(_VALID_HASH_A) else 3
- )
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex(_VALID_HASH_A) else 3
await manager.check_and_update_propagation_node()
@@ -92,9 +86,7 @@ async def test_auto_propagation_logic():
context=context,
)
- config.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
- _VALID_HASH_C
- )
+ config.lxmf_preferred_propagation_node_destination_hash.get.return_value = _VALID_HASH_C
announce3 = {
"destination_hash": _VALID_HASH_C,
"app_data": _APP_DATA_ENABLED,
@@ -108,9 +100,7 @@ async def test_auto_propagation_logic():
patch.object(manager, "_wait_for_path", return_value=True),
patch.object(manager, "_probe_propagation_sync", return_value=True),
):
- mock_hops.side_effect = lambda dh: (
- 1 if dh == bytes.fromhex(_VALID_HASH_A) else 2
- )
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex(_VALID_HASH_A) else 2
await manager.check_and_update_propagation_node()
diff --git a/tests/backend/test_auto_propagation_api.py b/tests/backend/test_auto_propagation_api.py
index 9b5d11b8..692cbe8a 100644
--- a/tests/backend/test_auto_propagation_api.py
+++ b/tests/backend/test_auto_propagation_api.py
@@ -103,9 +103,7 @@ async def test_auto_propagation_api(mock_rns_minimal, temp_dir):
response = await patch_handler(mock_request)
data = json.loads(response.body)
assert data["config"]["lxmf_preferred_propagation_node_auto_select"] is False
- assert (
- app_instance.config.lxmf_preferred_propagation_node_auto_select.get() is False
- )
+ assert app_instance.config.lxmf_preferred_propagation_node_auto_select.get() is False
# Update transfer/sync limits and validate clamping/application
mock_request = MagicMock()
@@ -133,7 +131,5 @@ async def test_auto_propagation_api(mock_rns_minimal, temp_dir):
response = await patch_handler(mock_request)
data = json.loads(response.body)
assert data["config"]["lxmf_delivery_transfer_limit_in_bytes"] == 1_000_000_000
- assert (
- app_instance.config.lxmf_delivery_transfer_limit_in_bytes.get() == 1_000_000_000
- )
+ assert app_instance.config.lxmf_delivery_transfer_limit_in_bytes.get() == 1_000_000_000
assert app_instance.message_router.delivery_per_transfer_limit == 1_000_000
diff --git a/tests/backend/test_bot_handler_extended.py b/tests/backend/test_bot_handler_extended.py
index de27e253..09fafc56 100644
--- a/tests/backend/test_bot_handler_extended.py
+++ b/tests/backend/test_bot_handler_extended.py
@@ -111,9 +111,7 @@ def test_get_status_reads_sidecar_lxmf_address(temp_identity_dir):
storage = os.path.join(handler.bots_dir, sid)
os.makedirs(storage, exist_ok=True)
hx = "a" * 32
- with open(
- os.path.join(storage, "meshchatx_lxmf_address.txt"), "w", encoding="utf-8"
- ) as f:
+ with open(os.path.join(storage, "meshchatx_lxmf_address.txt"), "w", encoding="utf-8") as f:
f.write(hx)
handler.bots_state = [{"id": sid, "template_id": "echo", "storage_dir": storage}]
status = handler.get_status()
@@ -178,9 +176,7 @@ def test_request_announce_writes_trigger(mock_alive, temp_identity_dir):
sid = "b1"
storage = os.path.join(handler.bots_dir, sid)
os.makedirs(storage, exist_ok=True)
- handler.bots_state = [
- {"id": sid, "template_id": "echo", "storage_dir": storage, "pid": 99999}
- ]
+ handler.bots_state = [{"id": sid, "template_id": "echo", "storage_dir": storage, "pid": 99999}]
handler.request_announce(sid)
req = os.path.join(storage, "meshchatx_request_announce")
assert os.path.isfile(req)
@@ -193,8 +189,6 @@ def test_request_announce_not_running(temp_identity_dir):
sid = "b1"
storage = os.path.join(handler.bots_dir, sid)
os.makedirs(storage, exist_ok=True)
- handler.bots_state = [
- {"id": sid, "template_id": "echo", "storage_dir": storage, "pid": None}
- ]
+ handler.bots_state = [{"id": sid, "template_id": "echo", "storage_dir": storage, "pid": None}]
with pytest.raises(RuntimeError, match="not running"):
handler.request_announce(sid)
diff --git a/tests/backend/test_concurrency_stress.py b/tests/backend/test_concurrency_stress.py
index 71e2cd3c..f0cf0bc4 100644
--- a/tests/backend/test_concurrency_stress.py
+++ b/tests/backend/test_concurrency_stress.py
@@ -91,12 +91,8 @@ class TestConcurrencyStress(unittest.TestCase):
def test_database_concurrency(self):
"""Launches multiple reader and writer threads to check for lock contention."""
- writers = [
- threading.Thread(target=self.db_writer_worker, args=(i,)) for i in range(5)
- ]
- readers = [
- threading.Thread(target=self.db_reader_worker, args=(i,)) for i in range(5)
- ]
+ writers = [threading.Thread(target=self.db_writer_worker, args=(i,)) for i in range(5)]
+ readers = [threading.Thread(target=self.db_reader_worker, args=(i,)) for i in range(5)]
for t in writers + readers:
t.start()
diff --git a/tests/backend/test_contacts_display_name_semantics.py b/tests/backend/test_contacts_display_name_semantics.py
index ccdb74e3..9f38c861 100644
--- a/tests/backend/test_contacts_display_name_semantics.py
+++ b/tests/backend/test_contacts_display_name_semantics.py
@@ -176,9 +176,7 @@ class TestCustomDisplayNameLifecycle:
def test_unicode_display_name(self, announce_dao):
announce_dao.upsert_custom_display_name("dest1", "\u5c71\u7530\u592a\u90ce")
- assert (
- announce_dao.get_custom_display_name("dest1") == "\u5c71\u7530\u592a\u90ce"
- )
+ assert announce_dao.get_custom_display_name("dest1") == "\u5c71\u7530\u592a\u90ce"
def test_very_long_display_name(self, announce_dao):
long_name = "A" * 10000
diff --git a/tests/backend/test_contacts_export_import.py b/tests/backend/test_contacts_export_import.py
index fd8fb7e6..d9e74c00 100644
--- a/tests/backend/test_contacts_export_import.py
+++ b/tests/backend/test_contacts_export_import.py
@@ -48,10 +48,7 @@ async def test_contacts_export_empty(mock_rns_minimal, temp_dir):
)
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/telephone/contacts/export"
- and route.method == "GET"
- ):
+ if route.path == "/api/v1/telephone/contacts/export" and route.method == "GET":
handler = route.handler
break
assert handler is not None
@@ -76,10 +73,7 @@ async def test_contacts_export_with_data(mock_rns_minimal, temp_dir):
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/telephone/contacts/export"
- and route.method == "GET"
- ):
+ if route.path == "/api/v1/telephone/contacts/export" and route.method == "GET":
handler = route.handler
break
assert handler is not None
@@ -107,10 +101,7 @@ async def test_contacts_import_valid(mock_rns_minimal, temp_dir):
)
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/telephone/contacts/import"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/telephone/contacts/import" and route.method == "POST":
handler = route.handler
break
assert handler is not None
@@ -147,10 +138,7 @@ async def test_contacts_import_skips_invalid(mock_rns_minimal, temp_dir):
)
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/telephone/contacts/import"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/telephone/contacts/import" and route.method == "POST":
handler = route.handler
break
assert handler is not None
@@ -181,10 +169,7 @@ async def test_contacts_import_rejects_non_array(mock_rns_minimal, temp_dir):
)
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/telephone/contacts/import"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/telephone/contacts/import" and route.method == "POST":
handler = route.handler
break
assert handler is not None
diff --git a/tests/backend/test_csp_logic.py b/tests/backend/test_csp_logic.py
index f660eda0..587c51f3 100644
--- a/tests/backend/test_csp_logic.py
+++ b/tests/backend/test_csp_logic.py
@@ -113,7 +113,6 @@ async def test_config_update_csp(mock_rns_minimal, tmp_path):
assert response.status == 200
assert (
- app_instance.config.csp_extra_connect_src.get()
- == "https://api1.com, https://api2.com"
+ app_instance.config.csp_extra_connect_src.get() == "https://api1.com, https://api2.com"
)
assert app_instance.config.csp_extra_img_src.get() == "https://img.com"
diff --git a/tests/backend/test_dao_fuzzing.py b/tests/backend/test_dao_fuzzing.py
index ae4c437a..909d4109 100644
--- a/tests/backend/test_dao_fuzzing.py
+++ b/tests/backend/test_dao_fuzzing.py
@@ -662,9 +662,7 @@ class TestSafeHrefFuzzing:
url = f"{scheme}:something"
result = _safe_href(url)
lower = url.strip().lower()
- if any(
- lower.startswith(p) for p in ("https://", "http://", "/", "#", "mailto:")
- ):
+ if any(lower.startswith(p) for p in ("https://", "http://", "/", "#", "mailto:")):
assert result == url
else:
assert result == "#"
diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py
index 8767e8f5..cdc285b2 100644
--- a/tests/backend/test_database_snapshots.py
+++ b/tests/backend/test_database_snapshots.py
@@ -139,9 +139,7 @@ def test_backup_suspicious_when_messages_gone_skips_cleanup_and_baseline(temp_di
assert "baseline" in result2
assert result2["baseline"]["message_count"] == 1
assert result2["current_stats"]["message_count"] == 0
- zip_count_after_suspicious = sum(
- 1 for f in os.listdir(backup_dir) if f.endswith(".zip")
- )
+ zip_count_after_suspicious = sum(1 for f in os.listdir(backup_dir) if f.endswith(".zip"))
assert zip_count_after_suspicious == 2
assert any("SUSPICIOUS" in f for f in os.listdir(backup_dir) if f.endswith(".zip"))
with open(os.path.join(backup_dir, "backup-baseline.json")) as f:
@@ -304,13 +302,8 @@ def test_is_backup_suspicious_does_not_mistrigger_empty_baseline():
db = Database(":memory:")
db.initialize()
- assert (
- db._is_backup_suspicious({"message_count": 0, "total_bytes": 0}, None) is False
- )
- assert (
- db._is_backup_suspicious({"message_count": 10, "total_bytes": 1000}, None)
- is False
- )
+ assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 0}, None) is False
+ assert db._is_backup_suspicious({"message_count": 10, "total_bytes": 1000}, None) is False
def test_is_backup_suspicious_does_not_mistrigger_legitimate_empty():
@@ -319,10 +312,7 @@ def test_is_backup_suspicious_does_not_mistrigger_legitimate_empty():
db = Database(":memory:")
db.initialize()
baseline = {"message_count": 0, "total_bytes": 5000}
- assert (
- db._is_backup_suspicious({"message_count": 0, "total_bytes": 5000}, baseline)
- is False
- )
+ assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 5000}, baseline) is False
def test_is_backup_suspicious_does_not_mistrigger_small_db():
@@ -331,7 +321,4 @@ def test_is_backup_suspicious_does_not_mistrigger_small_db():
db = Database(":memory:")
db.initialize()
baseline = {"message_count": 5, "total_bytes": 50_000}
- assert (
- db._is_backup_suspicious({"message_count": 5, "total_bytes": 55_000}, baseline)
- is False
- )
+ assert db._is_backup_suspicious({"message_count": 5, "total_bytes": 55_000}, baseline) is False
diff --git a/tests/backend/test_docs_manager.py b/tests/backend/test_docs_manager.py
index 9d27a364..77de751f 100644
--- a/tests/backend/test_docs_manager.py
+++ b/tests/backend/test_docs_manager.py
@@ -115,7 +115,7 @@ def test_docs_manager_readonly_public_dir_handling(tmp_path):
public_dir = tmp_path / "readonly_public"
public_dir.mkdir()
- os.chmod(public_dir, 0o555)
+ os.chmod(public_dir, 0o555) # noqa: S103
config = MagicMock()
from unittest.mock import patch
@@ -123,12 +123,9 @@ def test_docs_manager_readonly_public_dir_handling(tmp_path):
with patch("os.makedirs", side_effect=OSError("Read-only file system")):
dm = DocsManager(config, str(public_dir))
assert dm.last_error is not None
- assert (
- "Read-only file system" in dm.last_error
- or "Permission denied" in dm.last_error
- )
+ assert "Read-only file system" in dm.last_error or "Permission denied" in dm.last_error
- os.chmod(public_dir, 0o755)
+ os.chmod(public_dir, 0o755) # noqa: S103
def test_has_docs(docs_manager, temp_dirs):
diff --git a/tests/backend/test_fuzzing.py b/tests/backend/test_fuzzing.py
index 4b3c77ca..72df3a31 100644
--- a/tests/backend/test_fuzzing.py
+++ b/tests/backend/test_fuzzing.py
@@ -283,10 +283,10 @@ def mock_app(temp_dir):
app.config.auto_send_failed_messages_to_propagation_node.get.return_value = True
app.config.show_suggested_community_interfaces.get.return_value = True
app.config.lxmf_local_propagation_node_enabled.get.return_value = False
- app.config.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
- None
+ app.config.lxmf_preferred_propagation_node_destination_hash.get.return_value = None
+ app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get.return_value = (
+ 3600
)
- app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get.return_value = 3600
app.config.lxmf_preferred_propagation_node_last_synced_at.get.return_value = 0
app.config.lxmf_user_icon_name.get.return_value = "user"
app.config.lxmf_user_icon_foreground_colour.get.return_value = "#ffffff"
@@ -297,16 +297,10 @@ def mock_app(temp_dir):
app.config.lxmf_auto_sync_propagation_nodes_min_hops.get.return_value = 1
app.config.lxmf_auto_sync_propagation_nodes_max_hops.get.return_value = 5
app.config.lxmf_auto_sync_propagation_nodes_max_count.get.return_value = 10
- app.config.lxmf_auto_sync_propagation_nodes_max_age_seconds.get.return_value = (
- 86400
- )
- app.config.lxmf_auto_sync_propagation_nodes_max_size_bytes.get.return_value = (
- 1000000
- )
+ app.config.lxmf_auto_sync_propagation_nodes_max_age_seconds.get.return_value = 86400
+ app.config.lxmf_auto_sync_propagation_nodes_max_size_bytes.get.return_value = 1000000
app.config.lxmf_auto_sync_propagation_nodes_max_total_size_bytes.get.return_value = 10000000
- app.config.lxmf_auto_sync_propagation_nodes_max_total_count.get.return_value = (
- 100
- )
+ app.config.lxmf_auto_sync_propagation_nodes_max_total_count.get.return_value = 100
app.config.lxmf_auto_sync_propagation_nodes_max_total_age_seconds.get.return_value = 864000
app.config.lxmf_auto_sync_propagation_nodes_max_total_size_bytes_per_node.get.return_value = 1000000
app.config.lxmf_auto_sync_propagation_nodes_max_total_count_per_node.get.return_value = 100
diff --git a/tests/backend/test_identity_switch.py b/tests/backend/test_identity_switch.py
index 77864f0f..799e22e7 100644
--- a/tests/backend/test_identity_switch.py
+++ b/tests/backend/test_identity_switch.py
@@ -59,9 +59,7 @@ def mock_rns():
# Apply patches
mocks = {}
for p in patches:
- attr_name = (
- p.attribute if hasattr(p, "attribute") else p.target.split(".")[-1]
- )
+ attr_name = p.attribute if hasattr(p, "attribute") else p.target.split(".")[-1]
mocks[attr_name] = stack.enter_context(p)
# Mock class methods on MockIdentityClass
diff --git a/tests/backend/test_integrity.py b/tests/backend/test_integrity.py
index 07b69786..9e987679 100644
--- a/tests/backend/test_integrity.py
+++ b/tests/backend/test_integrity.py
@@ -87,8 +87,7 @@ class TestIntegrityManager(unittest.TestCase):
self.assertFalse(is_ok)
self.assertTrue(
any(
- "Critical security component" in i or "File signature mismatch" in i
- for i in issues
+ "Critical security component" in i or "File signature mismatch" in i for i in issues
),
)
diff --git a/tests/backend/test_interface_discovery.py b/tests/backend/test_interface_discovery.py
index b950ff6c..a4e34c3d 100644
--- a/tests/backend/test_interface_discovery.py
+++ b/tests/backend/test_interface_discovery.py
@@ -100,9 +100,7 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
assert get_data["discovery"]["discover_interfaces"] == "true"
assert get_data["discovery"]["interface_discovery_sources"] == "abc,def"
assert get_data["discovery"]["interface_discovery_whitelist"] == "tcp-*,10.0.*"
- assert (
- get_data["discovery"]["interface_discovery_blacklist"] == "tcp-bad,*:9999"
- )
+ assert get_data["discovery"]["interface_discovery_blacklist"] == "tcp-bad,*:9999"
assert get_data["discovery"]["required_discovery_value"] == "16"
assert get_data["discovery"]["autoconnect_discovered_interfaces"] == "2"
assert get_data["discovery"]["network_identity"] == "/tmp/net_id"
@@ -127,10 +125,7 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
patch_data = json.loads(patch_response.body)
assert patch_data["discovery"]["discover_interfaces"] is False
assert patch_data["discovery"]["interface_discovery_sources"] is None
- assert (
- patch_data["discovery"]["interface_discovery_whitelist"]
- == "peer-*,172.16.*"
- )
+ assert patch_data["discovery"]["interface_discovery_whitelist"] == "peer-*,172.16.*"
assert patch_data["discovery"]["interface_discovery_blacklist"] is None
assert patch_data["discovery"]["required_discovery_value"] == 18
assert patch_data["discovery"]["autoconnect_discovered_interfaces"] == 5
@@ -263,15 +258,9 @@ async def test_discovery_patch_sanitizes_whitelist_blacklist_values(temp_dir):
data = json.loads(response.body)
assert data["discovery"]["interface_discovery_whitelist"] == "peer-1,host:4242"
- assert (
- data["discovery"]["interface_discovery_blacklist"] == "bad-node,evilentry"
- )
- assert (
- config["reticulum"]["interface_discovery_whitelist"] == "peer-1,host:4242"
- )
- assert (
- config["reticulum"]["interface_discovery_blacklist"] == "bad-node,evilentry"
- )
+ assert data["discovery"]["interface_discovery_blacklist"] == "bad-node,evilentry"
+ assert config["reticulum"]["interface_discovery_whitelist"] == "peer-1,host:4242"
+ assert config["reticulum"]["interface_discovery_blacklist"] == "bad-node,evilentry"
assert config.write_called
@@ -477,8 +466,8 @@ async def test_interface_add_discovery_payload_fuzz_tcp_client(temp_dir):
class AddRequest:
@staticmethod
- async def json():
- return payload
+ async def json(p=payload):
+ return p
response = await add_handler(AddRequest())
data = json.loads(response.body)
diff --git a/tests/backend/test_interface_discovery_ifac.py b/tests/backend/test_interface_discovery_ifac.py
index 27b8fc7b..56b8f645 100644
--- a/tests/backend/test_interface_discovery_ifac.py
+++ b/tests/backend/test_interface_discovery_ifac.py
@@ -61,9 +61,7 @@ async def find_route_handler(app_instance, path, method):
def test_normalize_handles_non_list_input():
- assert ReticulumMeshChat.normalize_discovered_ifac_fields({"foo": "bar"}) == {
- "foo": "bar"
- }
+ assert ReticulumMeshChat.normalize_discovered_ifac_fields({"foo": "bar"}) == {"foo": "bar"}
assert ReticulumMeshChat.normalize_discovered_ifac_fields(None) is None
diff --git a/tests/backend/test_interface_port_check.py b/tests/backend/test_interface_port_check.py
index 3c897f31..10077baf 100644
--- a/tests/backend/test_interface_port_check.py
+++ b/tests/backend/test_interface_port_check.py
@@ -51,12 +51,12 @@ def test_is_port_in_use_rejects_invalid_inputs(port):
def test_is_port_in_use_handles_wildcard_host():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.bind(("0.0.0.0", 0))
+ sock.bind(("0.0.0.0", 0)) # noqa: S104
sock.listen(1)
port = sock.getsockname()[1]
try:
assert is_port_in_use(None, port, kind="tcp") is True
- assert is_port_in_use("0.0.0.0", port, kind="tcp") is True
+ assert is_port_in_use("0.0.0.0", port, kind="tcp") is True # noqa: S104
finally:
sock.close()
diff --git a/tests/backend/test_licenses_collector.py b/tests/backend/test_licenses_collector.py
index 05e5ecd2..235a66c4 100644
--- a/tests/backend/test_licenses_collector.py
+++ b/tests/backend/test_licenses_collector.py
@@ -102,12 +102,8 @@ def test_build_licenses_payload_composes_counts_and_meta():
def test_render_third_party_notices_contains_sections_and_rows():
payload = {
- "backend": [
- {"name": "rns", "version": "1.0", "author": "Author A", "license": "MIT"}
- ],
- "frontend": [
- {"name": "vue", "version": "3.0", "author": "Author B", "license": "MIT"}
- ],
+ "backend": [{"name": "rns", "version": "1.0", "author": "Author A", "license": "MIT"}],
+ "frontend": [{"name": "vue", "version": "3.0", "author": "Author B", "license": "MIT"}],
"meta": {"generated_at": "2026-01-01T00:00:00Z", "frontend_source": "pnpm"},
}
rendered = render_third_party_notices(payload)
@@ -135,9 +131,7 @@ def test_write_embedded_license_artifacts_writes_files(tmp_path):
assert frontend_path.exists()
assert notices_path.exists()
assert '"name": "vue"' in frontend_path.read_text(encoding="utf-8")
- assert "Reticulum MeshChatX - Third-party notices" in notices_path.read_text(
- encoding="utf-8"
- )
+ assert "Reticulum MeshChatX - Third-party notices" in notices_path.read_text(encoding="utf-8")
def test_write_embedded_license_artifacts_preserves_existing_frontend_when_empty(
diff --git a/tests/backend/test_lxmf_attachments.py b/tests/backend/test_lxmf_attachments.py
index 38518522..0a6a0ce7 100644
--- a/tests/backend/test_lxmf_attachments.py
+++ b/tests/backend/test_lxmf_attachments.py
@@ -18,9 +18,7 @@ def test_message_fields_have_attachments():
assert message_fields_have_attachments(json.dumps({"audio": "base64data"})) is True
# File attachments - empty list
- assert (
- message_fields_have_attachments(json.dumps({"file_attachments": []})) is False
- )
+ assert message_fields_have_attachments(json.dumps({"file_attachments": []})) is False
# File attachments - with files
assert (
diff --git a/tests/backend/test_lxmf_communication.py b/tests/backend/test_lxmf_communication.py
index 4a286190..a01aea16 100644
--- a/tests/backend/test_lxmf_communication.py
+++ b/tests/backend/test_lxmf_communication.py
@@ -20,9 +20,8 @@ import sys
import textwrap
import pytest
-
-import LXMF.LXStamper as LXStamper
import RNS
+from LXMF import LXStamper
_RUN = os.environ.get("MESHCHAT_LIVE_RETICULUM") == "1"
@@ -37,13 +36,13 @@ _MINIMAL_RNS_CONFIG = """\
def _run_lxmf_script(script_body, timeout=120):
- result = subprocess.run(
+ return subprocess.run(
[sys.executable, "-c", script_body],
capture_output=True,
text=True,
timeout=timeout,
+ check=False,
)
- return result
def _parse_result(proc):
diff --git a/tests/backend/test_lxmf_icons.py b/tests/backend/test_lxmf_icons.py
index 13c386f1..416b49f3 100644
--- a/tests/backend/test_lxmf_icons.py
+++ b/tests/backend/test_lxmf_icons.py
@@ -61,9 +61,7 @@ def mock_rns():
# Apply patches
mocks = {}
for p in patches:
- attr_name = (
- p.attribute if hasattr(p, "attribute") else p.target.split(".")[-1]
- )
+ attr_name = p.attribute if hasattr(p, "attribute") else p.target.split(".")[-1]
mocks[attr_name] = stack.enter_context(p)
# Access specifically the ones we need to configure
@@ -72,13 +70,11 @@ def mock_rns():
# Setup mock config
mock_config.return_value.display_name.get.return_value = "Test User"
mock_config.return_value.lxmf_user_icon_name.get.return_value = "user"
- mock_config.return_value.lxmf_user_icon_foreground_colour.get.return_value = (
- "#ffffff"
+ mock_config.return_value.lxmf_user_icon_foreground_colour.get.return_value = "#ffffff"
+ mock_config.return_value.lxmf_user_icon_background_colour.get.return_value = "#000000"
+ mock_config.return_value.auto_send_failed_messages_to_propagation_node.get.return_value = (
+ False
)
- mock_config.return_value.lxmf_user_icon_background_colour.get.return_value = (
- "#000000"
- )
- mock_config.return_value.auto_send_failed_messages_to_propagation_node.get.return_value = False
# Mock class methods on MockIdentityClass
mock_id_instance = MockIdentityClass()
diff --git a/tests/backend/test_lxmf_propagation_full.py b/tests/backend/test_lxmf_propagation_full.py
index 7db609a2..5db70fbb 100644
--- a/tests/backend/test_lxmf_propagation_full.py
+++ b/tests/backend/test_lxmf_propagation_full.py
@@ -104,10 +104,7 @@ async def test_lxmf_propagation_config(mock_app):
mock_app.current_context.message_router.set_outbound_propagation_node.assert_called_with(
node_hash_bytes,
)
- assert (
- mock_app.config.lxmf_preferred_propagation_node_destination_hash.get()
- == node_hash_hex
- )
+ assert mock_app.config.lxmf_preferred_propagation_node_destination_hash.get() == node_hash_hex
@pytest.mark.asyncio
@@ -128,9 +125,7 @@ async def test_lxmf_sync_flow(mock_app):
mock_router.propagation_transfer_progress = 0.75
status_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/status"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/status"
)
response = await status_handler(None)
data = json.loads(response.body)
@@ -144,9 +139,7 @@ async def test_lxmf_sync_requests_path_before_sync(mock_app):
outbound = b"somehash"
mock_router.get_outbound_propagation_node.return_value = outbound
sync_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/sync"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/sync"
)
with patch("meshchatx.meshchat.RNS.Transport.has_path", return_value=False):
@@ -165,9 +158,7 @@ async def test_lxmf_sync_completes_immediately_for_local_preferred_node(mock_app
mock_router.propagation_destination = SimpleNamespace(hash=local_hash)
mock_router.get_outbound_propagation_node.return_value = local_hash
sync_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/sync"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/sync"
)
await sync_handler(None)
@@ -212,19 +203,14 @@ async def test_auto_sync_interval_config(mock_app):
await mock_app.update_config(
{"lxmf_preferred_propagation_node_auto_sync_interval_seconds": 3600},
)
- assert (
- mock_app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
- == 3600
- )
+ assert mock_app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get() == 3600
@pytest.mark.asyncio
async def test_propagation_node_status_mapping(mock_app):
mock_router = mock_app.current_context.message_router
status_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/status"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/status"
)
states_to_test = [
@@ -271,9 +257,7 @@ async def test_local_propagation_node_stop_and_restart_routes(mock_app):
}
stop_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/stop"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/stop"
)
restart_handler = next(
r.handler
@@ -318,13 +302,13 @@ async def test_user_provided_node_hash(mock_app):
)
# Trigger a sync request
- mock_app.current_context.message_router.get_outbound_propagation_node.return_value = bytes.fromhex(
- node_hash_hex,
+ mock_app.current_context.message_router.get_outbound_propagation_node.return_value = (
+ bytes.fromhex(
+ node_hash_hex,
+ )
)
sync_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/sync"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/sync"
)
await sync_handler(None)
diff --git a/tests/backend/test_lxmf_propagation_sync_integration.py b/tests/backend/test_lxmf_propagation_sync_integration.py
index f702e4f9..cc5065d3 100644
--- a/tests/backend/test_lxmf_propagation_sync_integration.py
+++ b/tests/backend/test_lxmf_propagation_sync_integration.py
@@ -104,9 +104,7 @@ def integration_app(temp_dir):
def _route_handler(app, path, method="GET"):
- return next(
- r.handler for r in app.get_routes() if r.path == path and r.method == method
- )
+ return next(r.handler for r in app.get_routes() if r.path == path and r.method == method)
@pytest.mark.asyncio
@@ -135,24 +133,18 @@ async def test_remote_propagation_sync_transitions_path_requested_to_complete(
with (
patch("meshchatx.meshchat.RNS.Transport.has_path", side_effect=has_path),
- patch(
- "meshchatx.meshchat.RNS.Transport.request_path", side_effect=request_path
- ),
+ patch("meshchatx.meshchat.RNS.Transport.request_path", side_effect=request_path),
):
first_sync = await sync_handler(None)
assert first_sync.status == 200
- first_status = json.loads((await status_handler(None)).body)[
- "propagation_node_status"
- ]
+ first_status = json.loads((await status_handler(None)).body)["propagation_node_status"]
assert first_status["state"] == "path_requested"
second_sync = await sync_handler(None)
assert second_sync.status == 200
- second_status = json.loads((await status_handler(None)).body)[
- "propagation_node_status"
- ]
+ second_status = json.loads((await status_handler(None)).body)["propagation_node_status"]
assert second_status["state"] == "complete"
assert second_status["progress"] == 100.0
assert fake_router.request_messages_calls >= 2
@@ -177,9 +169,7 @@ async def test_local_preferred_propagation_sync_completes_without_remote_lookup(
response = await sync_handler(None)
assert response.status == 200
- status_data = json.loads((await status_handler(None)).body)[
- "propagation_node_status"
- ]
+ status_data = json.loads((await status_handler(None)).body)["propagation_node_status"]
assert status_data["state"] == "complete"
assert status_data["progress"] == 100.0
assert fake_router.request_messages_calls == 0
diff --git a/tests/backend/test_lxmf_sync.py b/tests/backend/test_lxmf_sync.py
index 32166566..743a92e9 100644
--- a/tests/backend/test_lxmf_sync.py
+++ b/tests/backend/test_lxmf_sync.py
@@ -78,10 +78,7 @@ async def test_lxmf_sync_endpoints(mock_app):
# 1. Test status endpoint initially idle
handler = None
for route in mock_app.get_routes():
- if (
- route.path == "/api/v1/lxmf/propagation-node/status"
- and route.method == "GET"
- ):
+ if route.path == "/api/v1/lxmf/propagation-node/status" and route.method == "GET":
handler = route.handler
break
@@ -107,9 +104,7 @@ async def test_lxmf_sync_endpoints(mock_app):
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_once()
# 3. Test status change to complete
- mock_app.current_context.message_router.propagation_transfer_state = (
- LXMF.LXMRouter.PR_COMPLETE
- )
+ mock_app.current_context.message_router.propagation_transfer_state = LXMF.LXMRouter.PR_COMPLETE
response = await handler(None)
data = json.loads(response.body)
assert data["propagation_node_status"]["state"] == "complete"
@@ -140,7 +135,9 @@ async def test_specific_node_hash_validation(mock_app):
break
# Ensure it's considered configured
- mock_app.current_context.message_router.get_outbound_propagation_node.return_value = expected_bytes
+ mock_app.current_context.message_router.get_outbound_propagation_node.return_value = (
+ expected_bytes
+ )
await sync_handler(None)
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_once()
@@ -149,14 +146,10 @@ async def test_specific_node_hash_validation(mock_app):
@pytest.mark.asyncio
async def test_status_includes_sync_storage_and_confirmation_metrics(mock_app):
status_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/status"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/status"
)
sync_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/sync"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/sync"
)
mock_app.current_context.message_router.get_outbound_propagation_node.return_value = b"somehash"
@@ -187,9 +180,7 @@ async def test_status_includes_sync_storage_and_confirmation_metrics(mock_app):
@pytest.mark.asyncio
async def test_status_metrics_default_to_zero_before_any_sync(mock_app):
status_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/status"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/status"
)
response = await status_handler(None)
data = json.loads(response.body)["propagation_node_status"]
@@ -203,14 +194,10 @@ async def test_status_metrics_default_to_zero_before_any_sync(mock_app):
@pytest.mark.asyncio
async def test_status_hidden_metric_is_clamped_to_zero(mock_app):
status_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/status"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/status"
)
sync_handler = next(
- r.handler
- for r in mock_app.get_routes()
- if r.path == "/api/v1/lxmf/propagation-node/sync"
+ r.handler for r in mock_app.get_routes() if r.path == "/api/v1/lxmf/propagation-node/sync"
)
mock_app.current_context.message_router.get_outbound_propagation_node.return_value = b"somehash"
diff --git a/tests/backend/test_lxmf_utils_extended.py b/tests/backend/test_lxmf_utils_extended.py
index ea2c1683..fb4b1adf 100644
--- a/tests/backend/test_lxmf_utils_extended.py
+++ b/tests/backend/test_lxmf_utils_extended.py
@@ -76,15 +76,9 @@ def test_convert_lxmf_message_to_dict_with_attachments():
== base64.b64encode(b"content1").decode()
)
assert result["fields"]["image"]["image_type"] == "png"
- assert (
- result["fields"]["image"]["image_bytes"]
- == base64.b64encode(b"image_data").decode()
- )
+ assert result["fields"]["image"]["image_bytes"] == base64.b64encode(b"image_data").decode()
assert result["fields"]["audio"]["audio_mode"] == "voice"
- assert (
- result["fields"]["audio"]["audio_bytes"]
- == base64.b64encode(b"audio_data").decode()
- )
+ assert result["fields"]["audio"]["audio_bytes"] == base64.b64encode(b"audio_data").decode()
def test_convert_lxmf_state_to_string():
diff --git a/tests/backend/test_markdown_renderer.py b/tests/backend/test_markdown_renderer.py
index 8014d279..c608aae0 100644
--- a/tests/backend/test_markdown_renderer.py
+++ b/tests/backend/test_markdown_renderer.py
@@ -26,8 +26,7 @@ class TestMarkdownRenderer(unittest.TestCase):
self.assertIn("<code", rendered)
self.assertIn("language-python", rendered)
self.assertTrue(
- "print('hello')" in rendered
- or "print('hello')" in rendered,
+ "print('hello')" in rendered or "print('hello')" in rendered,
)
def test_lists(self):
diff --git a/tests/backend/test_media_fuzzing.py b/tests/backend/test_media_fuzzing.py
index 93fa9ca8..bb24c416 100644
--- a/tests/backend/test_media_fuzzing.py
+++ b/tests/backend/test_media_fuzzing.py
@@ -62,9 +62,7 @@ def test_parse_tgs_gzip_json_fuzz(payload):
merged.setdefault("op", 60.0)
merged.setdefault("w", 100)
merged.setdefault("h", 100)
- raw = gzip.compress(
- json.dumps(merged, default=str).encode("utf-8", errors="surrogateescape")
- )
+ raw = gzip.compress(json.dumps(merged, default=str).encode("utf-8", errors="surrogateescape"))
if len(raw) > sticker_utils.MAX_ANIMATED_BYTES:
raw = raw[: sticker_utils.MAX_ANIMATED_BYTES]
try:
@@ -118,9 +116,7 @@ def test_extract_metadata_fuzz_never_raises(image_type, raw):
typ=st.one_of(
st.none(),
st.text(max_size=48),
- st.sampled_from(
- ["png", "jpeg", "jpg", "webp", "gif", "bmp", "tgs", "webm", "svg", ""]
- ),
+ st.sampled_from(["png", "jpeg", "jpg", "webp", "gif", "bmp", "tgs", "webm", "svg", ""]),
),
strict=st.booleans(),
)
@@ -165,9 +161,7 @@ def test_mime_for_image_type_fuzz_never_raises(t):
description=st.one_of(st.none(), st.text(max_size=400)),
pack_type=st.one_of(st.none(), st.text(max_size=40)),
)
-def test_sticker_pack_sanitizers_fuzz_never_raises(
- title, short_name, description, pack_type
-):
+def test_sticker_pack_sanitizers_fuzz_never_raises(title, short_name, description, pack_type):
sticker_pack_utils.sanitize_pack_title(title)
sticker_pack_utils.sanitize_pack_short_name(short_name)
sticker_pack_utils.sanitize_pack_description(description)
diff --git a/tests/backend/test_meshchat_coverage.py b/tests/backend/test_meshchat_coverage.py
index 31bc97df..c747537f 100644
--- a/tests/backend/test_meshchat_coverage.py
+++ b/tests/backend/test_meshchat_coverage.py
@@ -452,9 +452,7 @@ def _build_wav_pcm16(samplerate=48000, duration_seconds=0.5, frequency=440.0):
wf.setframerate(samplerate)
frames = bytearray()
for i in range(n_samples):
- sample = int(
- 0.3 * 32767 * math.sin(2 * math.pi * frequency * (i / samplerate))
- )
+ sample = int(0.3 * 32767 * math.sin(2 * math.pi * frequency * (i / samplerate)))
frames.extend(struct.pack("<h", sample))
wf.writeframes(bytes(frames))
return buf.getvalue()
diff --git a/tests/backend/test_meshchat_utils.py b/tests/backend/test_meshchat_utils.py
index 7cea8eac..5e8ec597 100644
--- a/tests/backend/test_meshchat_utils.py
+++ b/tests/backend/test_meshchat_utils.py
@@ -104,12 +104,11 @@ def mock_app(temp_dir):
patch.object(MockIdentityClass, "from_bytes", return_value=mock_id),
)
- app = ReticulumMeshChat(
+ return ReticulumMeshChat(
identity=mock_id,
storage_dir=temp_dir,
reticulum_config_dir=temp_dir,
)
- return app
def test_get_interfaces_snapshot(mock_app):
diff --git a/tests/backend/test_message_sending_failures.py b/tests/backend/test_message_sending_failures.py
index 69fdbd6d..55ce1a21 100644
--- a/tests/backend/test_message_sending_failures.py
+++ b/tests/backend/test_message_sending_failures.py
@@ -2,7 +2,7 @@
import asyncio
import json
-from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
+from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
import LXMF
import pytest
diff --git a/tests/backend/test_notification_user_facing_filter.py b/tests/backend/test_notification_user_facing_filter.py
index 4e56987f..c8addd60 100644
--- a/tests/backend/test_notification_user_facing_filter.py
+++ b/tests/backend/test_notification_user_facing_filter.py
@@ -32,7 +32,6 @@ from meshchatx.src.backend.lxmf_utils import (
)
from meshchatx.src.backend.message_handler import MessageHandler
-
LOCAL_HASH = "aa" * 16
PEER_HASH = "bb" * 16
PEER_HASH_2 = "cc" * 16
diff --git a/tests/backend/test_package_version_resolution.py b/tests/backend/test_package_version_resolution.py
index 13c0b848..dcd56d88 100644
--- a/tests/backend/test_package_version_resolution.py
+++ b/tests/backend/test_package_version_resolution.py
@@ -84,9 +84,7 @@ def test_app_info_dependency_keys_resolve_in_dev_env(package: str):
assert v != "unknown", f"{package} must resolve when installed"
-@pytest.mark.skipif(
- sys.version_info < (3, 13), reason="audioop-lts only on Python 3.13+"
-)
+@pytest.mark.skipif(sys.version_info < (3, 13), reason="audioop-lts only on Python 3.13+")
def test_audioop_lts_resolves_when_applicable():
v = ReticulumMeshChat.get_package_version("audioop-lts")
assert v != "unknown"
diff --git a/tests/backend/test_performance_bottlenecks.py b/tests/backend/test_performance_bottlenecks.py
index bee75b8f..c1f41206 100644
--- a/tests/backend/test_performance_bottlenecks.py
+++ b/tests/backend/test_performance_bottlenecks.py
@@ -196,9 +196,7 @@ class TestPerformanceBottlenecks(unittest.TestCase):
b"packet",
)
- threads = [
- threading.Thread(target=insert_announces) for _ in range(num_threads)
- ]
+ threads = [threading.Thread(target=insert_announces) for _ in range(num_threads)]
print(
f"\nRunning {num_threads} threads inserting {announces_per_thread} announces each...",
diff --git a/tests/backend/test_performance_hotpaths.py b/tests/backend/test_performance_hotpaths.py
index baf71708..073e7942 100644
--- a/tests/backend/test_performance_hotpaths.py
+++ b/tests/backend/test_performance_hotpaths.py
@@ -512,9 +512,7 @@ class TestPerformanceHotPaths(unittest.TestCase):
with lock:
all_durations.extend(thread_durations)
- threads = [
- threading.Thread(target=writer, args=(t,)) for t in range(num_threads)
- ]
+ threads = [threading.Thread(target=writer, args=(t,)) for t in range(num_threads)]
t0 = time.perf_counter()
for t in threads:
t.start()
@@ -558,9 +556,7 @@ class TestPerformanceHotPaths(unittest.TestCase):
with lock:
all_durations.extend(thread_durations)
- threads = [
- threading.Thread(target=writer, args=(t,)) for t in range(num_threads)
- ]
+ threads = [threading.Thread(target=writer, args=(t,)) for t in range(num_threads)]
t0 = time.perf_counter()
for t in threads:
t.start()
@@ -623,12 +619,8 @@ class TestPerformanceHotPaths(unittest.TestCase):
with lock:
read_durations.extend(local_durs)
- writers = [
- threading.Thread(target=writer, args=(t,)) for t in range(num_writers)
- ]
- readers = [
- threading.Thread(target=reader, args=(t,)) for t in range(num_readers)
- ]
+ writers = [threading.Thread(target=writer, args=(t,)) for t in range(num_writers)]
+ readers = [threading.Thread(target=reader, args=(t,)) for t in range(num_readers)]
t0 = time.perf_counter()
for t in writers + readers:
diff --git a/tests/backend/test_propagation_nodes_robustness.py b/tests/backend/test_propagation_nodes_robustness.py
index db53b26d..e79834a1 100644
--- a/tests/backend/test_propagation_nodes_robustness.py
+++ b/tests/backend/test_propagation_nodes_robustness.py
@@ -67,10 +67,7 @@ async def test_propagation_nodes_endpoint_robustness(mock_rns_minimal, temp_dir)
assert "is_local_node" in node
if node.get("is_local_node") and isinstance(node.get("local_node_stats"), dict):
if isinstance(node["local_node_stats"].get("is_running"), bool):
- assert (
- node.get("is_propagation_enabled")
- == node["local_node_stats"]["is_running"]
- )
+ assert node.get("is_propagation_enabled") == node["local_node_stats"]["is_running"]
# Test with invalid limit (should not crash)
request.query = {"limit": "invalid"}
@@ -82,10 +79,7 @@ async def test_propagation_nodes_endpoint_robustness(mock_rns_minimal, temp_dir)
assert "is_local_node" in node
if node.get("is_local_node") and isinstance(node.get("local_node_stats"), dict):
if isinstance(node["local_node_stats"].get("is_running"), bool):
- assert (
- node.get("is_propagation_enabled")
- == node["local_node_stats"]["is_running"]
- )
+ assert node.get("is_propagation_enabled") == node["local_node_stats"]["is_running"]
# Test with missing limit (should not crash)
request.query = {}
@@ -97,7 +91,4 @@ async def test_propagation_nodes_endpoint_robustness(mock_rns_minimal, temp_dir)
assert "is_local_node" in node
if node.get("is_local_node") and isinstance(node.get("local_node_stats"), dict):
if isinstance(node["local_node_stats"].get("is_running"), bool):
- assert (
- node.get("is_propagation_enabled")
- == node["local_node_stats"]["is_running"]
- )
+ assert node.get("is_propagation_enabled") == node["local_node_stats"]["is_running"]
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index 4f56e53d..0d65ebbf 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -213,7 +213,7 @@ def test_interface_config_parser_best_effort_property(names, keys, values):
config_lines = ["[interfaces]"]
for name in names:
config_lines.append(f"[[{name}") # Missing closing ]]
- for k, v in zip(keys, values):
+ for k, v in zip(keys, values, strict=False):
config_lines.append(f" {k} = {v}")
config_text = "\n".join(config_lines)
@@ -302,7 +302,7 @@ def test_interface_config_parser_structured(names, keys, values):
config_lines = ["[interfaces]"]
for name in names:
config_lines.append(f"[[{name}]]")
- for k, v in zip(keys, values):
+ for k, v in zip(keys, values, strict=False):
config_lines.append(f" {k} = {v}")
config_text = "\n".join(config_lines)
diff --git a/tests/backend/test_reticulum_live_network.py b/tests/backend/test_reticulum_live_network.py
index 43d5dd7e..69784dfc 100644
--- a/tests/backend/test_reticulum_live_network.py
+++ b/tests/backend/test_reticulum_live_network.py
@@ -42,5 +42,6 @@ finally:
capture_output=True,
text=True,
timeout=120,
+ check=False,
)
assert result.returncode == 0, result.stderr + result.stdout
diff --git a/tests/backend/test_rnode_download_firmware.py b/tests/backend/test_rnode_download_firmware.py
index ccf3a00d..0a845ce8 100644
--- a/tests/backend/test_rnode_download_firmware.py
+++ b/tests/backend/test_rnode_download_firmware.py
@@ -98,9 +98,7 @@ async def test_download_firmware_returns_zip_for_allowed_url(web_app):
async with TestClient(TestServer(aio_app)) as client:
r = await client.get(
"/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip"
- },
+ params={"url": "https://github.com/owner/repo/releases/download/v1/firmware.zip"},
)
assert r.status == 200
assert r.headers.get("Content-Type", "").startswith("application/zip")
@@ -144,9 +142,7 @@ async def test_download_firmware_returns_500_on_exception(web_app):
async with TestClient(TestServer(aio_app)) as client:
r = await client.get(
"/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip"
- },
+ params={"url": "https://github.com/owner/repo/releases/download/v1/firmware.zip"},
)
assert r.status == 500
body = await r.json()
diff --git a/tests/backend/test_rnpath_logic.py b/tests/backend/test_rnpath_logic.py
index b2ae8c41..9ba15164 100644
--- a/tests/backend/test_rnpath_logic.py
+++ b/tests/backend/test_rnpath_logic.py
@@ -70,9 +70,7 @@ async def test_rnpath_table_endpoint(mock_rns_minimal, temp_dir):
request.query = {}
handler = next(
- r.handler
- for r in app_instance.get_routes()
- if r.path == "/api/v1/rnpath/table"
+ r.handler for r in app_instance.get_routes() if r.path == "/api/v1/rnpath/table"
)
response = await handler(request)
data = json.loads(response.body)
@@ -98,9 +96,7 @@ async def test_rnpath_request_endpoint(mock_rns_minimal, temp_dir):
request.json = AsyncMock(return_value={"destination_hash": target_hash})
handler = next(
- r.handler
- for r in app_instance.get_routes()
- if r.path == "/api/v1/rnpath/request"
+ r.handler for r in app_instance.get_routes() if r.path == "/api/v1/rnpath/request"
)
response = await handler(request)
@@ -122,9 +118,7 @@ async def test_rnpath_drop_endpoint(mock_rns_minimal, temp_dir):
request.json = AsyncMock(return_value={"destination_hash": target_hash})
handler = next(
- r.handler
- for r in app_instance.get_routes()
- if r.path == "/api/v1/rnpath/drop"
+ r.handler for r in app_instance.get_routes() if r.path == "/api/v1/rnpath/drop"
)
response = await handler(request)
diff --git a/tests/backend/test_rns_lifecycle.py b/tests/backend/test_rns_lifecycle.py
index 9b90c52d..060969c0 100644
--- a/tests/backend/test_rns_lifecycle.py
+++ b/tests/backend/test_rns_lifecycle.py
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: 0BSD
-import os
import json
+import os
import shutil
import socket
import tempfile
@@ -528,10 +528,7 @@ async def test_transport_enable_endpoint_reloads_rns(mock_rns, temp_dir):
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/reticulum/enable-transport"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/reticulum/enable-transport" and route.method == "POST":
handler = route.handler
break
@@ -541,10 +538,7 @@ async def test_transport_enable_endpoint_reloads_rns(mock_rns, temp_dir):
payload = json.loads(response.body)
assert response.status == 200
- assert (
- payload["message"]
- == "Transport mode enabled and RNS restarted successfully."
- )
+ assert payload["message"] == "Transport mode enabled and RNS restarted successfully."
app.reload_reticulum.assert_awaited_once()
app.teardown_identity()
@@ -576,10 +570,7 @@ async def test_transport_disable_endpoint_reloads_rns(mock_rns, temp_dir):
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/reticulum/disable-transport"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/reticulum/disable-transport" and route.method == "POST":
handler = route.handler
break
@@ -589,10 +580,7 @@ async def test_transport_disable_endpoint_reloads_rns(mock_rns, temp_dir):
payload = json.loads(response.body)
assert response.status == 200
- assert (
- payload["message"]
- == "Transport mode disabled and RNS restarted successfully."
- )
+ assert payload["message"] == "Transport mode disabled and RNS restarted successfully."
app.reload_reticulum.assert_awaited_once()
app.teardown_identity()
@@ -624,10 +612,7 @@ async def test_transport_enable_endpoint_reload_failure(mock_rns, temp_dir):
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/reticulum/enable-transport"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/reticulum/enable-transport" and route.method == "POST":
handler = route.handler
break
@@ -637,10 +622,7 @@ async def test_transport_enable_endpoint_reload_failure(mock_rns, temp_dir):
payload = json.loads(response.body)
assert response.status == 500
- assert (
- payload["message"]
- == "Transport mode was enabled in config, but RNS reload failed."
- )
+ assert payload["message"] == "Transport mode was enabled in config, but RNS reload failed."
app.reload_reticulum.assert_awaited_once()
app.teardown_identity()
@@ -672,10 +654,7 @@ async def test_transport_disable_endpoint_reload_failure(mock_rns, temp_dir):
handler = None
for route in app.get_routes():
- if (
- route.path == "/api/v1/reticulum/disable-transport"
- and route.method == "POST"
- ):
+ if route.path == "/api/v1/reticulum/disable-transport" and route.method == "POST":
handler = route.handler
break
@@ -685,10 +664,7 @@ async def test_transport_disable_endpoint_reload_failure(mock_rns, temp_dir):
payload = json.loads(response.body)
assert response.status == 500
- assert (
- payload["message"]
- == "Transport mode was disabled in config, but RNS reload failed."
- )
+ assert payload["message"] == "Transport mode was disabled in config, but RNS reload failed."
app.reload_reticulum.assert_awaited_once()
app.teardown_identity()
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index 0eeebbba..f7eba93e 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: 0BSD
import base64
+import math
import os
import time
from contextlib import ExitStack
@@ -646,9 +647,7 @@ def test_lxm_uri_comprehensive_fuzzing(mock_app, uri):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
- uri_str = (
- uri.decode("utf-8", errors="ignore") if isinstance(uri, bytes) else uri
- )
+ uri_str = uri.decode("utf-8", errors="ignore") if isinstance(uri, bytes) else uri
loop.run_until_complete(
mock_app.on_websocket_data_received(
mock_client,
@@ -1123,20 +1122,17 @@ def test_map_tile_coordinates_fuzzing(mock_app, z, x, y):
try:
z_int = (
int(z)
- if isinstance(z, (int, float))
- and not (isinstance(z, float) and (z != z or abs(z) == float("inf")))
+ if isinstance(z, (int, float)) and (not isinstance(z, float) or math.isfinite(z))
else 0
)
x_int = (
int(x)
- if isinstance(x, (int, float))
- and not (isinstance(x, float) and (x != x or abs(x) == float("inf")))
+ if isinstance(x, (int, float)) and (not isinstance(x, float) or math.isfinite(x))
else 0
)
y_int = (
int(y)
- if isinstance(y, (int, float))
- and not (isinstance(y, float) and (y != y or abs(y) == float("inf")))
+ if isinstance(y, (int, float)) and (not isinstance(y, float) or math.isfinite(y))
else 0
)
mock_app.map_manager.get_tile(z_int, x_int, y_int)
@@ -1403,9 +1399,7 @@ def test_nomadnet_page_archive_add_fuzzing(
import asyncio
content_str = (
- content.decode("utf-8", errors="replace")
- if isinstance(content, bytes)
- else content
+ content.decode("utf-8", errors="replace") if isinstance(content, bytes) else content
)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
@@ -1942,10 +1936,7 @@ def test_lxmf_audio_mode_fuzzing(mock_app, audio_mode, audio_bytes):
)
def test_lxst_profile_switching_fuzzing(mock_app, profile_id):
"""Fuzz LXST audio profile switching."""
- if (
- hasattr(mock_app.telephone_manager, "telephone")
- and mock_app.telephone_manager.telephone
- ):
+ if hasattr(mock_app.telephone_manager, "telephone") and mock_app.telephone_manager.telephone:
mock_app.telephone_manager.telephone.switch_profile(profile_id)
@@ -1979,10 +1970,7 @@ def test_lxst_call_initiation_fuzzing(mock_app, destination_hash, timeout):
timeout_int = (
int(timeout)
if isinstance(timeout, (int, float))
- and not (
- isinstance(timeout, float)
- and (timeout != timeout or abs(timeout) == float("inf"))
- )
+ and (not isinstance(timeout, float) or math.isfinite(timeout))
else 15
)
@@ -2188,15 +2176,11 @@ def test_lxmf_display_name_parsing_regression():
# None case (fallback to default)
mock_parser.return_value = None
- assert (
- parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
- )
+ assert parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
# Exception case
mock_parser.side_effect = Exception("Parsing error")
- assert (
- parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
- )
+ assert parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback"
# None input
assert parse_lxmf_display_name(None, default_value="Fallback") == "Fallback"
diff --git a/tests/backend/test_smoke_extended.py b/tests/backend/test_smoke_extended.py
index a3705335..24361972 100644
--- a/tests/backend/test_smoke_extended.py
+++ b/tests/backend/test_smoke_extended.py
@@ -15,6 +15,7 @@ def test_cli_help():
[sys.executable, "-m", "meshchatx.meshchat", "--help"],
capture_output=True,
text=True,
+ check=False,
)
assert result.returncode == 0
assert "usage:" in result.stdout.lower() or "options:" in result.stdout.lower()
diff --git a/tests/backend/test_startup.py b/tests/backend/test_startup.py
index b44b46a0..b02de103 100644
--- a/tests/backend/test_startup.py
+++ b/tests/backend/test_startup.py
@@ -114,18 +114,14 @@ def test_reticulum_meshchat_init(mock_rns, temp_dir):
# Setup config mock values
mock_config_instance.auth_enabled.get.return_value = False
mock_config_instance.lxmf_propagation_node_stamp_cost.get.return_value = 0
- mock_config_instance.lxmf_delivery_transfer_limit_in_bytes.get.return_value = (
- 1000000
- )
+ mock_config_instance.lxmf_delivery_transfer_limit_in_bytes.get.return_value = 1000000
mock_config_instance.lxmf_inbound_stamp_cost.get.return_value = 0
mock_config_instance.display_name.get.return_value = "Test User"
- mock_config_instance.lxmf_preferred_propagation_node_destination_hash.get.return_value = None
- mock_config_instance.lxmf_local_propagation_node_enabled.get.return_value = (
- False
- )
- mock_config_instance.libretranslate_url.get.return_value = (
- "http://localhost:5000"
+ mock_config_instance.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
+ None
)
+ mock_config_instance.lxmf_local_propagation_node_enabled.get.return_value = False
+ mock_config_instance.libretranslate_url.get.return_value = "http://localhost:5000"
mock_config_instance.translator_enabled.get.return_value = False
app = ReticulumMeshChat(
diff --git a/tests/backend/test_sticker_pack_utils.py b/tests/backend/test_sticker_pack_utils.py
index 8b1f1cc2..7d68aadb 100644
--- a/tests/backend/test_sticker_pack_utils.py
+++ b/tests/backend/test_sticker_pack_utils.py
@@ -19,10 +19,7 @@ def test_sanitize_pack_title_default():
def test_sanitize_pack_short_name():
assert sticker_pack_utils.sanitize_pack_short_name(None) is None
assert sticker_pack_utils.sanitize_pack_short_name("Cats!@#") == "cats"
- assert (
- sticker_pack_utils.sanitize_pack_short_name(" Hello_World-1 ")
- == "hello_world-1"
- )
+ assert sticker_pack_utils.sanitize_pack_short_name(" Hello_World-1 ") == "hello_world-1"
assert sticker_pack_utils.sanitize_pack_short_name("***") is None
assert len(sticker_pack_utils.sanitize_pack_short_name("a" * 200)) == 32
diff --git a/tests/backend/test_sticker_utils.py b/tests/backend/test_sticker_utils.py
index 6d25f402..edfdfb42 100644
--- a/tests/backend/test_sticker_utils.py
+++ b/tests/backend/test_sticker_utils.py
@@ -59,24 +59,13 @@ def test_validate_sticker_payload_magic_type_mismatch():
def test_detect_image_format_from_magic():
assert sticker_utils.detect_image_format_from_magic(b"\x89PNG\r\n\x1a\n") == "png"
- assert (
- sticker_utils.detect_image_format_from_magic(b"\xff\xd8\xff\xe0\x00\x10")
- == "jpeg"
- )
- assert (
- sticker_utils.detect_image_format_from_magic(b"GIF89a" + b"\x00" * 4) == "gif"
- )
+ assert sticker_utils.detect_image_format_from_magic(b"\xff\xd8\xff\xe0\x00\x10") == "jpeg"
+ assert sticker_utils.detect_image_format_from_magic(b"GIF89a" + b"\x00" * 4) == "gif"
assert sticker_utils.detect_image_format_from_magic(b"BM" + b"\x00" * 20) == "bmp"
webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 8
assert sticker_utils.detect_image_format_from_magic(webp) == "webp"
- assert (
- sticker_utils.detect_image_format_from_magic(b"\x1a\x45\xdf\xa3" + b"\x00" * 8)
- == "webm"
- )
- assert (
- sticker_utils.detect_image_format_from_magic(b"\x1f\x8b\x08\x00" + b"\x00" * 8)
- == "tgs"
- )
+ assert sticker_utils.detect_image_format_from_magic(b"\x1a\x45\xdf\xa3" + b"\x00" * 8) == "webm"
+ assert sticker_utils.detect_image_format_from_magic(b"\x1f\x8b\x08\x00" + b"\x00" * 8) == "tgs"
assert sticker_utils.detect_image_format_from_magic(b"") is None
assert sticker_utils.detect_image_format_from_magic(b"short") is None
@@ -199,9 +188,7 @@ def test_validate_export_document_fuzz_never_raises_unexpected(doc):
pass
-def _build_tgs(
- width: int = 512, height: int = 512, fps: float = 30.0, frames: int = 60
-) -> bytes:
+def _build_tgs(width: int = 512, height: int = 512, fps: float = 30.0, frames: int = 60) -> bytes:
import gzip
import json
@@ -247,9 +234,7 @@ def test_parse_tgs_invalid_metadata():
import gzip
import json
- raw = gzip.compress(
- json.dumps({"w": 0, "h": 0, "fr": 0, "ip": 0, "op": 0}).encode()
- )
+ raw = gzip.compress(json.dumps({"w": 0, "h": 0, "fr": 0, "ip": 0, "op": 0}).encode())
with pytest.raises(ValueError, match="invalid_tgs_metadata"):
sticker_utils.parse_tgs(raw)
@@ -305,12 +290,7 @@ def test_validate_strict_webm_too_large():
def test_detect_image_dimensions_png():
- raw = (
- b"\x89PNG\r\n\x1a\n"
- + b"\x00\x00\x00\rIHDR"
- + struct_pack(512, 512)
- + b"\x08\x06\x00\x00\x00"
- )
+ raw = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR" + struct_pack(512, 512) + b"\x08\x06\x00\x00\x00"
assert sticker_utils.detect_image_dimensions("png", raw) == (512, 512)
@@ -349,12 +329,7 @@ def test_extract_metadata_tgs():
def test_extract_metadata_static_png():
- raw = (
- b"\x89PNG\r\n\x1a\n"
- + b"\x00\x00\x00\rIHDR"
- + struct_pack(512, 256)
- + b"\x08\x06\x00\x00\x00"
- )
+ raw = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR" + struct_pack(512, 256) + b"\x08\x06\x00\x00\x00"
meta = sticker_utils.extract_metadata("png", raw)
assert meta["width"] == 512
assert meta["height"] == 256
diff --git a/tests/backend/test_telemetry_integration.py b/tests/backend/test_telemetry_integration.py
index 85ea40ec..0ef2fb84 100644
--- a/tests/backend/test_telemetry_integration.py
+++ b/tests/backend/test_telemetry_integration.py
@@ -36,8 +36,8 @@ def mock_app():
# Attach the actual method we want to test if possible,
# but since it's an instance method, we might need to bind it.
- app.process_incoming_telemetry = (
- ReticulumMeshChat.process_incoming_telemetry.__get__(app, ReticulumMeshChat)
+ app.process_incoming_telemetry = ReticulumMeshChat.process_incoming_telemetry.__get__(
+ app, ReticulumMeshChat
)
return app
diff --git a/tests/backend/test_telephone_api_json_contracts.py b/tests/backend/test_telephone_api_json_contracts.py
index f95f3eda..baebb4db 100644
--- a/tests/backend/test_telephone_api_json_contracts.py
+++ b/tests/backend/test_telephone_api_json_contracts.py
@@ -73,18 +73,14 @@ class _Query:
@pytest.mark.asyncio
-async def test_api_v1_telephone_voicemail_status_json_contract(
- mock_rns_minimal, temp_dir
-):
+async def test_api_v1_telephone_voicemail_status_json_contract(mock_rns_minimal, temp_dir):
with patch("meshchatx.meshchat.generate_ssl_certificate"):
app_instance = ReticulumMeshChat(
identity=mock_rns_minimal,
storage_dir=temp_dir,
reticulum_config_dir=temp_dir,
)
- handler = _find_handler(
- app_instance, "/api/v1/telephone/voicemail/status", "GET"
- )
+ handler = _find_handler(app_instance, "/api/v1/telephone/voicemail/status", "GET")
assert handler is not None
request = MagicMock()
response = await handler(request)
@@ -110,9 +106,7 @@ async def test_api_v1_telephone_voicemails_json_contract(mock_rns_minimal, temp_
@pytest.mark.asyncio
-async def test_api_v1_telephone_ringtones_list_json_contract(
- mock_rns_minimal, temp_dir
-):
+async def test_api_v1_telephone_ringtones_list_json_contract(mock_rns_minimal, temp_dir):
with patch("meshchatx.meshchat.generate_ssl_certificate"):
app_instance = ReticulumMeshChat(
identity=mock_rns_minimal,
@@ -128,18 +122,14 @@ async def test_api_v1_telephone_ringtones_list_json_contract(
@pytest.mark.asyncio
-async def test_api_v1_telephone_ringtones_status_json_contract(
- mock_rns_minimal, temp_dir
-):
+async def test_api_v1_telephone_ringtones_status_json_contract(mock_rns_minimal, temp_dir):
with patch("meshchatx.meshchat.generate_ssl_certificate"):
app_instance = ReticulumMeshChat(
identity=mock_rns_minimal,
storage_dir=temp_dir,
reticulum_config_dir=temp_dir,
)
- handler = _find_handler(
- app_instance, "/api/v1/telephone/ringtones/status", "GET"
- )
+ handler = _find_handler(app_instance, "/api/v1/telephone/ringtones/status", "GET")
assert handler is not None
request = MagicMock()
request.query = _Query({})
@@ -166,9 +156,7 @@ async def test_api_v1_telephone_contacts_list_json_contract(mock_rns_minimal, te
@pytest.mark.asyncio
-async def test_api_v1_telephone_contacts_check_json_contract(
- mock_rns_minimal, temp_dir
-):
+async def test_api_v1_telephone_contacts_check_json_contract(mock_rns_minimal, temp_dir):
with patch("meshchatx.meshchat.generate_ssl_certificate"):
app_instance = ReticulumMeshChat(
identity=mock_rns_minimal,
diff --git a/tests/backend/test_telephone_initiation.py b/tests/backend/test_telephone_initiation.py
index 7ff00088..f2d24a54 100644
--- a/tests/backend/test_telephone_initiation.py
+++ b/tests/backend/test_telephone_initiation.py
@@ -1,8 +1,8 @@
# SPDX-License-Identifier: 0BSD
import asyncio
-import tracemalloc
import time
+import tracemalloc
from unittest.mock import MagicMock, patch
import pytest
@@ -21,9 +21,7 @@ def telephone_manager():
tm._path_retry_interval_s = 0.01
tm._status_poll_interval_s = 0.01
tm._status_events = []
- tm.on_initiation_status_callback = lambda status, _target: tm._status_events.append(
- status
- )
+ tm.on_initiation_status_callback = lambda status, _target: tm._status_events.append(status)
return tm
@@ -50,9 +48,7 @@ async def test_initiate_retries_path_requests_during_lookup(telephone_manager):
"meshchatx.src.backend.telephone_manager.RNS.Transport.has_path",
side_effect=has_path,
),
- patch(
- "meshchatx.src.backend.telephone_manager.RNS.Transport.request_path"
- ) as request_path,
+ patch("meshchatx.src.backend.telephone_manager.RNS.Transport.request_path") as request_path,
):
await telephone_manager.initiate(destination_hash, timeout_seconds=1)
@@ -83,9 +79,7 @@ async def test_initiate_cancels_quickly_while_finding_path_identity(telephone_ma
side_effect=request_path_and_cancel,
),
):
- task = asyncio.create_task(
- telephone_manager.initiate(destination_hash, timeout_seconds=5)
- )
+ task = asyncio.create_task(telephone_manager.initiate(destination_hash, timeout_seconds=5))
result = await asyncio.wait_for(task, timeout=0.3)
assert result is None
@@ -111,9 +105,7 @@ async def test_initiate_cancels_quickly_while_dialling(telephone_manager):
return_value=True,
),
):
- task = asyncio.create_task(
- telephone_manager.initiate(destination_hash, timeout_seconds=5)
- )
+ task = asyncio.create_task(telephone_manager.initiate(destination_hash, timeout_seconds=5))
for _ in range(200):
if telephone_manager.initiation_status in (
"Establishing link...",
@@ -178,9 +170,7 @@ async def test_cancel_after_path_found_before_dialling_stabilizes(telephone_mana
return_value=True,
),
):
- task = asyncio.create_task(
- telephone_manager.initiate(destination_hash, timeout_seconds=2)
- )
+ task = asyncio.create_task(telephone_manager.initiate(destination_hash, timeout_seconds=2))
for _ in range(200):
if telephone_manager.initiation_status == "Establishing link...":
break
@@ -285,9 +275,7 @@ async def test_call_thread_exception_surfaces_without_hanging(telephone_manager)
"meshchatx.src.backend.telephone_manager.RNS.Transport.has_path",
return_value=True,
),
- patch(
- "meshchatx.src.backend.telephone_manager.asyncio.sleep", side_effect=no_wait
- ),
+ patch("meshchatx.src.backend.telephone_manager.asyncio.sleep", side_effect=no_wait),
):
result = await asyncio.wait_for(
telephone_manager.initiate(destination_hash, timeout_seconds=1),
@@ -304,7 +292,6 @@ async def test_inconsistent_call_status_finishes_within_timeout(telephone_manage
def inconsistent_call(_identity):
telephone_manager.telephone.call_status = 5
- return None
telephone_manager.telephone.call.side_effect = inconsistent_call
@@ -320,9 +307,7 @@ async def test_inconsistent_call_status_finishes_within_timeout(telephone_manage
"meshchatx.src.backend.telephone_manager.RNS.Transport.has_path",
return_value=True,
),
- patch(
- "meshchatx.src.backend.telephone_manager.asyncio.sleep", side_effect=no_wait
- ),
+ patch("meshchatx.src.backend.telephone_manager.asyncio.sleep", side_effect=no_wait),
):
result = await asyncio.wait_for(
telephone_manager.initiate(destination_hash, timeout_seconds=0.2),
@@ -375,10 +360,8 @@ async def test_lxst_busy_and_rejected_end_without_stuck_status(telephone_manager
for terminal_state in (0, 1):
telephone_manager._status_events.clear()
telephone_manager.telephone.call_status = 3
- telephone_manager.telephone.call.side_effect = (
- lambda _identity, state=terminal_state: setattr(
- telephone_manager.telephone, "call_status", state
- )
+ telephone_manager.telephone.call.side_effect = lambda _identity, state=terminal_state: (
+ setattr(telephone_manager.telephone, "call_status", state)
)
with (
diff --git a/tests/backend/test_telephone_manager_boost.py b/tests/backend/test_telephone_manager_boost.py
index 77732eb7..b873a6aa 100644
--- a/tests/backend/test_telephone_manager_boost.py
+++ b/tests/backend/test_telephone_manager_boost.py
@@ -42,16 +42,12 @@ def test_init_telephone(mock_tel_class, tel_manager):
@patch("meshchatx.src.backend.telephone_manager.Telephone")
-def test_init_telephone_applies_config_audio_profile(
- mock_tel_class, mock_identity, tmp_path
-):
+def test_init_telephone_applies_config_audio_profile(mock_tel_class, mock_identity, tmp_path):
storage_dir = tmp_path / "tel"
storage_dir.mkdir()
cfg = MagicMock()
cfg.telephone_audio_profile_id.get.return_value = 96
- tm = TelephoneManager(
- mock_identity, config_manager=cfg, storage_dir=str(storage_dir)
- )
+ tm = TelephoneManager(mock_identity, config_manager=cfg, storage_dir=str(storage_dir))
tm.init_telephone()
mock_tel_class.return_value.switch_profile.assert_called_with(96)
diff --git a/tests/backend/test_telephone_recorder.py b/tests/backend/test_telephone_recorder.py
index 32717c98..710ca1e0 100644
--- a/tests/backend/test_telephone_recorder.py
+++ b/tests/backend/test_telephone_recorder.py
@@ -27,8 +27,7 @@ def mock_config():
@pytest.fixture
def mock_db():
- db = MagicMock()
- return db
+ return MagicMock()
@pytest.fixture
diff --git a/tests/backend/test_user_guidance_autointerface.py b/tests/backend/test_user_guidance_autointerface.py
index 702d413d..29e50b0f 100644
--- a/tests/backend/test_user_guidance_autointerface.py
+++ b/tests/backend/test_user_guidance_autointerface.py
@@ -19,12 +19,8 @@ def _make_app(config_interfaces):
config={"interfaces": dict(config_interfaces)},
transport_enabled=lambda: True,
)
- app._get_interfaces_section = (
- ReticulumMeshChat._get_interfaces_section.__get__(app)
- )
- app._detect_failed_autointerfaces = (
- ReticulumMeshChat._detect_failed_autointerfaces.__get__(app)
- )
+ app._get_interfaces_section = ReticulumMeshChat._get_interfaces_section.__get__(app)
+ app._detect_failed_autointerfaces = ReticulumMeshChat._detect_failed_autointerfaces.__get__(app)
return app
@@ -106,9 +102,7 @@ def test_guidance_message_emitted_for_failed_autointerface():
)
app.config = MagicMock()
app.config.auto_announce_enabled.get.return_value = True
- app.build_user_guidance_messages = (
- ReticulumMeshChat.build_user_guidance_messages.__get__(app)
- )
+ app.build_user_guidance_messages = ReticulumMeshChat.build_user_guidance_messages.__get__(app)
with patch("meshchatx.meshchat.RNS.Transport") as transport:
transport.interfaces = []
@@ -131,9 +125,7 @@ def test_guidance_message_absent_when_autointerface_running():
)
app.config = MagicMock()
app.config.auto_announce_enabled.get.return_value = True
- app.build_user_guidance_messages = (
- ReticulumMeshChat.build_user_guidance_messages.__get__(app)
- )
+ app.build_user_guidance_messages = ReticulumMeshChat.build_user_guidance_messages.__get__(app)
with patch("meshchatx.meshchat.RNS.Transport") as transport:
transport.interfaces = [_FakeAutoInterface()]
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────